#include "JoinTests.h"
#include "FlashStickTestTypes.h"

#include <time.h>
#include <fstream>
#include <iostream>
#include <stdlib.h>
#include <sstream>
#include <string>
#include <math.h>
#include <stdio.h>
#include <list>


using namespace std;

#define MEMORY_SIZE_BYTES		1024 // 1KB
#define	SMALL_REL_SIZE_BYTES	10*MEMORY_SIZE_BYTES // 10KB
#define BIG_REL_SIZE_BYTES		5*SMALL_REL_SIZE_BYTES // 50KB

#define	REL_ROW_SIZE_DWRODS		4
#define	REL_ROW_SIZE_BYTES		16//REL_ROW_SIZE_DWRODS * sizeof(int)
#define PAGE_SIZE_BYTES			512.0f

int JoinTests::nBufferLimitedByteCount	= MEMORY_SIZE_BYTES; 
int JoinTests::nSmallRelByteCount		= SMALL_REL_SIZE_BYTES;
int JoinTests::nBigRelByteCount			= BIG_REL_SIZE_BYTES;

JoinTests::JoinTests(void) :m_pSortTests(NULL)
{

	m_psRel1Buffer = NULL;
	m_psRel2Buffer = NULL;

	m_pSortTests	= new CSortTests();

#ifdef __FLASH__
	GetLogicalDiskPhysicalInfo(L"\\\\.\\E:\\");
#else
		GetLogicalDiskPhysicalInfo(L"\\\\.\\C:\\");
#endif // __FLASH__

	// initialize random seed:
	srand ( static_cast<unsigned int>(time(NULL)) );

}

JoinTests::~JoinTests(void)
{
}


bool JoinTests::NestedLoopJoin(wstring wsRel1FileName, wstring wsRel2FileName, wstring wsResFileName, float& rfTimeSec,
							   int& rnMatchesNum, int& rnReadSectorsNum)
{
	SRel1*		psRel1Buffer	= NULL;
	SRel2*		psRel2Buffer	= NULL;
	sRelResult* psResultBuffer	= NULL;

	bool bRes = true;


	int nRel1IterationNumber = m_nRel1ByteCount / nBufferLimitedByteCount;
	int nRel2IterationNumber = m_nRel2ByteCount / nBufferLimitedByteCount;
	int nResIterationNumber = 0;
	
	int nResRowsNum = nBufferLimitedByteCount / sizeof(sRelResult);

	int nBufferRel1RowsNum = m_nRel1RowsNum / nRel1IterationNumber;
	int nBufferRel2RowsNum = m_nRel2RowsNum / nRel2IterationNumber;

	if (AllocateVirtualBuffer((void**)(&psRel1Buffer), nBufferLimitedByteCount) == true)
	{
		if (AllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount) == true)
		{
			if (AllocateVirtualBuffer((void**)(&psResultBuffer), nBufferLimitedByteCount) != true)
			{
				printf("ERROR: cannot allocate results buffer \n");
			}
			bRes = true;
		}
	}



	Disk* psFlashDiskRel1	= new Disk(wsRel1FileName, eLogical, eRead);
	Disk* psFlashDiskRel2	= new Disk(wsRel2FileName, eLogical, eRead);
	Disk* psFlashDiskRes	= new Disk(wsResFileName, eLogical, eWrite);

	DWORDLONG dwlResOffset = 0;
	int nResIndex = 0;

	
	if ( psFlashDiskRel1->OpenDisk() && psFlashDiskRel2->OpenDisk() && psFlashDiskRes->OpenDisk() )
	{

		if (psFlashDiskRel1->CreateOverlappedEvent() && psFlashDiskRel2->CreateOverlappedEvent() && 
			psFlashDiskRes->CreateOverlappedEvent() )
		{
			DWORD dwBytesRead			= 0;
			DWORDLONG dwlPrepareOffset	= 0;
			int nResultCounter			= 0;
			int nReadSectors			= 0;

			DWORD dwStart = timeGetTime();

			for (int i = 0; i < nRel1IterationNumber; ++i)
			{
				dwlPrepareOffset = nBufferLimitedByteCount*i; 
				nReadSectors += nBufferLimitedByteCount / 512;

				if (psFlashDiskRel1->DoOverlappedIO(psRel1Buffer, nBufferLimitedByteCount, dwlPrepareOffset))
				{
		//				cout << i << ". Read buffer 1" << endl;
					int j = 0;
					for (int j = 0; j < nRel2IterationNumber; ++j)
					{
						dwlPrepareOffset = (nBufferLimitedByteCount*j);
						nReadSectors += nBufferLimitedByteCount / 512;	
						if (psFlashDiskRel2->DoOverlappedIO(psRel2Buffer, nBufferLimitedByteCount, dwlPrepareOffset))
						{
							//	cout << j << ". Read buffer 2" << endl;
							for (int m = 0; m < nBufferRel1RowsNum; ++m)
							{
								for (int k = 0; k < nBufferRel2RowsNum; ++k)
								{
									if (psRel1Buffer[m].m_nc == psRel2Buffer[k].m_nc)
									{
										// There is a match - insert to the result buffer
										psResultBuffer[nResIndex].m_sRel1.m_na = psRel1Buffer[m].m_na;
										psResultBuffer[nResIndex].m_sRel1.m_na = psRel1Buffer[m].m_nb;
										psResultBuffer[nResIndex].m_sRel1.m_na = psRel1Buffer[m].m_nc;
										psResultBuffer[nResIndex].m_sRel1.m_na = psRel1Buffer[m].m_nd;

										psResultBuffer[nResIndex].m_sRel2.m_nc = psRel2Buffer[k].m_nc;
										psResultBuffer[nResIndex].m_sRel2.m_ne = psRel2Buffer[k].m_ne;
										psResultBuffer[nResIndex].m_sRel2.m_nf = psRel2Buffer[k].m_nf;
										psResultBuffer[nResIndex].m_sRel2.m_ng = psRel2Buffer[k].m_ng;

										nResultCounter++;
										nResIndex++;
										if ( nResRowsNum == nResIndex)
										{
											//// This the maximum buffer we have - must write results to the flash disk
											//if (false == psFlashDiskRes->DoOverlappedIO(psResultBuffer, nResRowsNum * sizeof(sRelResult), dwlResOffset))
											//{
											//	printf("ERROR: Cannot write the results to the flash disk \n");
											//	bRes = false;
											//	break;
											//}
											nResIterationNumber++;
											dwlResOffset += nResRowsNum * sizeof(sRelResult) * nResIterationNumber;
											nResIndex = 0;
										}
									}

								}
							}
						}
						else
						{
							cout << "error reading relation 2 - iteration " << j << endl;
							break;
						}

//						printf("Rel 2: iteration %d finished \n", j);
					}

				}
				else
				{
					cout << "error reading relation 1 - iteration " << i << endl;
					break;
				}
//				printf("Rel 1: iteration %d finished \n", i);

			}

			DWORD dwStop = timeGetTime();

			rfTimeSec += CalcTime(dwStart, dwStop);

			cout << "Number Of Rows in Result Join Relation " << nResultCounter << endl << endl;
			cout << "Number of Read Sectors are: " << nReadSectors << endl;
			cout << "Number of result rows are: " << nResultCounter << endl;
			cout << "Number of writes to the flash disk: " << nResIterationNumber << endl;

			DeAllocateVirtualBuffer((void**)(&psRel1Buffer), nBufferLimitedByteCount);
			DeAllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount);
			DeAllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount);
			

			psFlashDiskRel1->CloseDisk();
			psFlashDiskRel2->CloseDisk();
			psFlashDiskRes->CloseDisk();

			rnMatchesNum		= nResultCounter;
			rnReadSectorsNum	= nReadSectors;

		}
	}


	delete psFlashDiskRel1;
	delete psFlashDiskRel2;
	delete psFlashDiskRes;

	return bRes;
}



bool JoinTests::CreateRandomRelations(	float fColumnCDensity1, 
										float fColumnCDensity2, int nRel1RowsNum, int nRel2RowsNum, int nRel1ByteCount, 
										int nRel2ByteCount, wstring wsRel1FileNameFlash, wstring wsRel2FileNameFlash/*, 
										wstring wsRel1FileNameDisk, wstring wsRel2FileNameDisk*/)
{


	bool bRes = true;

	float fColumnADensity = 1.0f; // 100%
	float fColumnBDensity = 0.1f; // 1%
	float fColumnDDensity = 0.1f; // 0.1%

	float fColumnEDensity = 0.2f; // 100%
	float fColumnFDensity = 0.1f; // 1%
	float fColumnGDensity = 0.1f; // 10%

	bRes = AllocateVirtualBuffer((void**)(&m_psRel1Buffer), nRel1ByteCount) ;
	if (bRes)
	{
		AllocateVirtualBuffer((void**)(&m_psRel2Buffer), nRel2ByteCount) ;
	}
	else
	{
		printf("not\n");
	}

	// initialize random seed:
	srand ( static_cast<unsigned int>(time(NULL)) );

	// Create relation1

	if (NULL == m_psRel1Buffer)
	{
		printf("*****\n");
	}

	for (int i = 0; i < nRel1RowsNum; ++i)
	{
		// range - 1..64000
		m_psRel1Buffer[i].m_na = rand() %  (static_cast<int>(fColumnADensity * nRel1RowsNum)) + 1;

		// range - 200..300
		m_psRel1Buffer[i].m_nb = rand() % (static_cast<int>(fColumnBDensity * nRel1RowsNum)) + 200;
		// range - 1..1000
		m_psRel1Buffer[i].m_nc = rand() % (static_cast<int>(fColumnCDensity1 * nRel1RowsNum)) + 1;

		// range - 50..60
		m_psRel1Buffer[i].m_nd = rand() % (static_cast<int>(fColumnDDensity * nRel1RowsNum))  + 50;//static_cast<int>((nIterationNum / fColumnDDensity * nIterationNum)) + 1;

	}

	cout << "Created relation 1" << endl;


	// Create relation2
	// The row2 C column density will be different then in row1
	for (int i = 0; i < nRel2RowsNum; ++i)
	{
		
		float fd		= static_cast<float>(rand()) / RAND_MAX;
		int nResult		= static_cast<int>(fd * nRel2RowsNum);

		// range - 1..6400
		m_psRel2Buffer[i].m_nc = nResult+1;//rand() %  nRel2RowsNum/*(static_cast<int>(fColumnCDensity1 * nRel1RowsNum))*/ + 1;

		// range - 200..300
		m_psRel2Buffer[i].m_ne = rand() % (static_cast<int>(fColumnEDensity * nRel2RowsNum)) + 200;
		// range - 1..1000
		m_psRel2Buffer[i].m_nf = rand() % (static_cast<int>(fColumnFDensity * nRel2RowsNum)) + 1;//static_cast<int>((nIterationNum / fColumnCDensity * nIterationNum)) + 1;

		// range - 50..60
		m_psRel2Buffer[i].m_ng = rand() % (static_cast<int>(fColumnGDensity * nRel2RowsNum))  + 50;//static_cast<int>((nIterationNum / fColumnDDensity * nIterationNum)) + 1;

	}

	cout << "Created relation 2" << endl;

#ifdef __FLASH__
	// Write the relations to the flash

	Disk* psFlashDiskRel1 = new Disk(wsRel1FileNameFlash, eLogical, eWrite);
	psFlashDiskRel1->OpenDisk();
	psFlashDiskRel1->CreateOverlappedEvent();
	psFlashDiskRel1->DoOverlappedIO(m_psRel1Buffer, nRel1ByteCount);
	psFlashDiskRel1->CloseDisk();
	delete psFlashDiskRel1;

	cout << "Wrote relation 1 to flash" << endl;

	Disk* psFlashDiskRel2 = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	psFlashDiskRel2->OpenDisk();
	psFlashDiskRel2->CreateOverlappedEvent();
	psFlashDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount);
	psFlashDiskRel2->CloseDisk();
	delete psFlashDiskRel2;
	
	cout << "Wrote relation 2 to flash" << endl;
#else
	// Write the relations to the hard disk too

	Disk* psDiskRel1 = new Disk(wsRel1FileNameFlash, eLogical, eWrite);
	if (psDiskRel1->OpenDisk() == true)
	{
		if( psDiskRel1->CreateOverlappedEvent())
		{
			if (psDiskRel1->DoOverlappedIO(m_psRel1Buffer, nRel1ByteCount))
			{
				cout << "Wrote relation 1 to hard-disk" << endl;
			}
			psDiskRel1->CloseDisk();		
		}
	}
	delete psDiskRel1;

	Disk* psDiskRel2 = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	if (psDiskRel2->OpenDisk() == true)
	{
		if( psDiskRel2->CreateOverlappedEvent())
		{
			if (psDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount))
			{
					cout << "Wrote relation 2 to hard-disk" << endl;
			}
			psDiskRel2->CloseDisk();		
		}
	}
	delete psDiskRel2;
	
#endif // #ifdef __FLASH__

	// Free the buffers
	DeAllocateVirtualBuffer((void**)(&m_psRel1Buffer), nRel1ByteCount);
	DeAllocateVirtualBuffer((void**)(&m_psRel2Buffer), nRel2ByteCount);

	return true;

}


bool JoinTests::CreateRandomRelations2(	float fColumnCDensity1, 
										float fColumnCDensity2, int nRel1RowsNum, int nRel2RowsNum, int nRel1ByteCount, 
										int nRel2ByteCount, wstring wsRel1FileNameFlash, wstring wsRel2FileNameFlash,
										int nNumOfSortedBlocks)
{


	bool bRes = true;


	if (false == AllocateVirtualBuffer((void**)(&m_psRel2Buffer), nRel2ByteCount))
	{
		return false;
	}


	// Create relation2
	// The row2 C column density will be different then in row1
	for (int i = 0; i < nRel2RowsNum; ++i)
	{
		
		float fd		= static_cast<float>(rand()) / RAND_MAX;
		int nResult		= static_cast<int>(fd * nRel2RowsNum);

		// range - 1..6400
		m_psRel2Buffer[i].m_nc = nResult+1;//rand() %  nRel2RowsNum/*(static_cast<int>(fColumnCDensity1 * nRel1RowsNum))*/ + 1;

		// range - 200..300
		m_psRel2Buffer[i].m_ne = 0;
		// range - 1..1000
		m_psRel2Buffer[i].m_nf = 0;

		// range - 50..60
		m_psRel2Buffer[i].m_ng = 0;

	}

	cout << "Created relation 2" << endl;

	int* pnBufferIndices = (int*)malloc(nNumOfSortedBlocks*sizeof(int));

	for (int i = 0; i < nNumOfSortedBlocks-1; ++i)
	{
		float fd		= static_cast<float>(rand()) / RAND_MAX;
		int nResult		= static_cast<int>(fd * nRel2RowsNum);

		pnBufferIndices[i] = nResult;
	}

	pnBufferIndices[nNumOfSortedBlocks-1] = nRel2RowsNum;

	m_pSortTests->QuickSort2<int>(pnBufferIndices, 0, nNumOfSortedBlocks-1);

	int nFirstIndex			= 0;

	for (int i = 0; i < nNumOfSortedBlocks; ++i)
	{
		printf("Buffer %d = (%d, %d), size = %d\n", i, nFirstIndex, pnBufferIndices[i]-1, pnBufferIndices[i]-1-nFirstIndex);
		m_pSortTests->QuickSort2<SRel2>(&(m_psRel2Buffer[nFirstIndex]), 0, pnBufferIndices[i]-1-nFirstIndex);
		nFirstIndex = pnBufferIndices[i];
	}

#ifdef __FLASH__
	// Write the relations to the flash

	Disk* psFlashDiskRel2 = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	psFlashDiskRel2->OpenDisk();
	psFlashDiskRel2->CreateOverlappedEvent();
	psFlashDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount);
	psFlashDiskRel2->CloseDisk();
	delete psFlashDiskRel2;
	
	cout << "Wrote relation 2 to flash" << endl;
#else
	// Write the relations to the hard disk too

	Disk* psDiskRel2 = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	if (psDiskRel2->OpenDisk() == true)
	{
		if( psDiskRel2->CreateOverlappedEvent())
		{
			if (psDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount))
			{
					cout << "Wrote relation 2 to hard-disk" << endl;
			}
			psDiskRel2->CloseDisk();		
		}
	}
	delete psDiskRel2;
	
#endif // #ifdef __FLASH__

	// Free the buffers
	DeAllocateVirtualBuffer((void**)(&m_psRel1Buffer),nRel1ByteCount);
	DeAllocateVirtualBuffer((void**)(&m_psRel2Buffer),nRel2ByteCount);

	return true;

}


bool JoinTests::CreateAlmostSortedRelation(float fUnsortedRatio, int nRel2RowsNum, 
					int nRel2ByteCount, wstring wsRel2FileNameFlash)
{

	
	bool bRes = true;

	int nNumOfUnsortedElements = 10;//static_cast<int>(floor(fUnsortedRatio*nBufferLimitedByteCount));

	if (false == AllocateVirtualBuffer((void**)(&m_psRel2Buffer), nRel2ByteCount))
	{
		return false;
	}


	// Create relation2
	// The row2 C column density will be different then in row1
	for (int i = 0; i < nRel2RowsNum; ++i)
	{
		// range - 1..6400
		m_psRel2Buffer[i].m_nc = i+1;

		// range - 200..300
		m_psRel2Buffer[i].m_ne = 0;
		// range - 1..1000
		m_psRel2Buffer[i].m_nf = 0;

		// range - 50..60
		m_psRel2Buffer[i].m_ng = 0;

	}

	cout << "Created relation 2" << endl;

	int* pnBufferIndices = (int*)malloc(nNumOfUnsortedElements*sizeof(int));

	for (int i = 0; i < nNumOfUnsortedElements; ++i)
	{
		//float fd		= static_cast<float>(rand()) / RAND_MAX;
		//int nResult		= static_cast<int>(fd * nRel2RowsNum);

		//pnBufferIndices[i] = nResult;

		pnBufferIndices[i] = (i+1)*2;
	}

	for (int i = 1; i < nNumOfUnsortedElements; ++i)
	{
		SRel2 sTemp = m_psRel2Buffer[pnBufferIndices[i-1]];
		m_psRel2Buffer[pnBufferIndices[i-1]] = m_psRel2Buffer[pnBufferIndices[i]];
		m_psRel2Buffer[pnBufferIndices[i]] = sTemp;
	}

#ifdef __FLASH__
	// Write the relations to the flash

	Disk* psFlashDiskRel2 = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	psFlashDiskRel2->OpenDisk();
	psFlashDiskRel2->CreateOverlappedEvent();
	psFlashDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount);
	psFlashDiskRel2->CloseDisk();
	delete psFlashDiskRel2;
	
	cout << "Wrote relation 2 to flash" << endl;
#else
	// Write the relations to the hard disk too

	Disk* psDiskRel2 = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	if (psDiskRel2->OpenDisk() == true)
	{
		if( psDiskRel2->CreateOverlappedEvent())
		{
			if (psDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount))
			{
					cout << "Wrote relation 2 to hard-disk" << endl;
			}
			psDiskRel2->CloseDisk();		
		}
	}
	delete psDiskRel2;
	
#endif // #ifdef __FLASH__

	// Free the buffers
	//DeAllocateVirtualBuffer((void**)(&m_psRel1Buffer),nRel1ByteCount);
	DeAllocateVirtualBuffer((void**)(&m_psRel2Buffer),nRel2ByteCount);

	return true;

}

bool JoinTests::CreateRel2SortedParts(	int nNumberOfSortedParts, int nRel2RowsNum, int nRel2ByteCount, wstring wsRel2FileNameFlash)
{
	bool bRes = true;


	if (false == AllocateVirtualBuffer((void**)(&m_psRel2Buffer), nRel2ByteCount))
	{
		return false;
	}

	// initialize random seed:
	srand ( static_cast<unsigned int>(time(NULL)) );

	memset(m_psRel2Buffer, 0, nRel2ByteCount);

	float fColumnEDensity = 1.0f;

		// Create relation2
	// The row2 C column density will be different then in row1
	for (int i = 0; i < nRel2RowsNum; ++i)
	{
		
		float fd		= static_cast<float>(rand()) / RAND_MAX;
		int nResult		= static_cast<int>(fd * nRel2RowsNum);

		// range - 1..6400
		m_psRel2Buffer[i].m_nc = nResult+1;

		// range - 200..300
		m_psRel2Buffer[i].m_ne = 0;//rand() % (static_cast<int>(fColumnEDensity * nRel2RowsNum)) + 200;
		// range - 1..1000
		m_psRel2Buffer[i].m_nf = 0;//rand() % (static_cast<int>(fColumnFDensity * nRel2RowsNum)) + 1;//static_cast<int>((nIterationNum / fColumnCDensity * nIterationNum)) + 1;

		// range - 50..60
		m_psRel2Buffer[i].m_ng = 0;//rand() % (static_cast<int>(fColumnGDensity * nRel2RowsNum))  + 50;//static_cast<int>((nIterationNum / fColumnDDensity * nIterationNum)) + 1;

	}

	cout << "Created relation 2" << endl;

	int numOfRowsInASortedBlock = (nRel2RowsNum/nNumberOfSortedParts);

	for (int i = 0; i < nNumberOfSortedParts; ++i)
	{
		// Sort every block
		m_pSortTests->QuickSort1<SRel2>((&(m_psRel2Buffer[i*numOfRowsInASortedBlock])), 0, numOfRowsInASortedBlock-1);

		for (int j = 0; j <numOfRowsInASortedBlock; ++j)
		{
			if (m_psRel2Buffer[i*numOfRowsInASortedBlock+j].GetKey() >= nRel2RowsNum/sizeof(SRel2))
			{
				printf("%d\n", m_psRel2Buffer[i*numOfRowsInASortedBlock+j].GetKey());
			}
		}
	}
	//// Create relation2
	//int nIndex = 0;
	//int nNumOfElementsInARow = (nRel2RowsNum/nNumberOfSortedParts);
	//for (int i = 0; i < nNumberOfSortedParts; ++i)
	//{
	//	for (int j = 0; j < nNumOfElementsInARow; ++j)
	//	{
	//		m_psRel2Buffer[nIndex++].m_nc = j;
	//	}
	//}

	//cout << "Created relation 2" << endl;

#ifdef __FLASH__
	// Write the relations to the flash

	Disk* psFlashDiskRel2 = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	psFlashDiskRel2->OpenDisk();
	psFlashDiskRel2->CreateOverlappedEvent();
	psFlashDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount);
	psFlashDiskRel2->CloseDisk();
	delete psFlashDiskRel2;
	
	cout << "Wrote relation 2 to flash" << endl;
#else
	// Write the relations to the hard disk too

	Disk* psDiskRel2 = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	if (psDiskRel2->OpenDisk() == true)
	{
		if( psDiskRel2->CreateOverlappedEvent())
		{
			if (psDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount))
			{
					cout << "Wrote relation 2 to hard-disk" << endl;
			}
			psDiskRel2->CloseDisk();		
		}
	}
	delete psDiskRel2;
	
#endif // #ifdef __FLASH__

	// Free the buffers
	DeAllocateVirtualBuffer((void**)(&m_psRel2Buffer), nRel2ByteCount);

	return true;

}

int	JoinTests::GetRandNum(int nLow, int nHigh)
{
	int nRes = ((rand() % (nHigh-nLow)) + nLow);
	return nRes;
}

bool JoinTests::CreateRandomRelations1(float fSmallRelationDencity, int nNumOfJoinRows, int nBigRelRowsNum, 
									int nSmallRelRowsNum, int nBigRelByteCount, 
									int nSmallRelByteCount, wstring wsBigRelFileName, wstring wsSmallRelFileName)
{
	// TODO: change the method to be generic (tamplate)
	bool bRes = true;

	int nSmallRelHighLimit = static_cast<int>(fSmallRelationDencity * nSmallRelRowsNum*2);

	bRes = AllocateVirtualBuffer((void**)(&m_psRel1Buffer), nBigRelByteCount) ;
	if (bRes)
	{
		AllocateVirtualBuffer((void**)(&m_psRel2Buffer), nSmallRelByteCount) ;
	}

	bool bChosen = false;
	int nRel2Nums = 0; 
	int nRel1Nums = 0;
	// Assumption the relations fields default values are 0
	
	for (int i = 0; i < nSmallRelRowsNum; ++i)
	{
		// Choose the rows in the small relation that will be in the result

		int nRandNum = GetRandNum(1,100);
		if (nRandNum < static_cast<int>(100*fSmallRelationDencity))
		{
			bChosen = true;
		}
		else
		{
			bChosen = false;
		}

		if (true == bChosen)
		{
			nRel2Nums++;
			// For simplicity, the numbers are just in the range of dencity
			m_psRel2Buffer[i].m_nc = 	GetRandNum(1,nSmallRelHighLimit);
			printf("%d/%d. %d - %d\n", i, nSmallRelRowsNum, nRandNum, m_psRel2Buffer[i].m_nc);

			// Choose the k rows in the big relation to be in the join and fill them with values

			// TODO: change to be not k but 1..k
			int k = 0;
			while (k < nNumOfJoinRows)
			{
			
				int nChosenRow = GetRandNum(0, nBigRelRowsNum-1);

				if ( 0 == m_psRel1Buffer[nChosenRow].m_nc)
				{
					// get the value of your match
					m_psRel1Buffer[nChosenRow].m_nc  = m_psRel2Buffer[i].m_nc ;
					k++;
					nRel1Nums++;
				}
			}

		}
		else
		{
			// If this row was not chosen, we will pick a number that won't be in the join for sure
			m_psRel2Buffer[i].m_nc = 	GetRandNum(nSmallRelHighLimit+1,nSmallRelRowsNum);
		}

		


	}

	printf("Rel2 nums are %d, Rel1 nums are %d \n", nRel2Nums, nRel1Nums);

	cout << "Created relation 2" << endl;

	// Fill the rest of the big-relation rows with dummt values
	for (int i = 0; i < nBigRelRowsNum; ++i)
	{

		if ( 0 == m_psRel1Buffer[i].m_nc)
		{
			// Get a dummy value which is not in the small relation
			m_psRel1Buffer[i].m_nc = 	GetRandNum(nSmallRelRowsNum+1,nBigRelRowsNum);
		}
	}

	cout << "Created relation 1" << endl;

	// Write the relations to the disk

	Disk* psDiskRel1 = new Disk(wsBigRelFileName, eLogical, eWrite);
	psDiskRel1->OpenDisk();
	psDiskRel1->CreateOverlappedEvent();
	psDiskRel1->DoOverlappedIO(m_psRel1Buffer, nBigRelByteCount);
	psDiskRel1->CloseDisk();
	delete psDiskRel1;

	cout << "Wrote relation 1 to the disk" << endl;

	psDiskRel1 = new Disk(wsSmallRelFileName, eLogical, eWrite);
	psDiskRel1->OpenDisk();
	psDiskRel1->CreateOverlappedEvent();
	psDiskRel1->DoOverlappedIO(m_psRel2Buffer, nSmallRelByteCount);
	psDiskRel1->CloseDisk();
	delete psDiskRel1;
	
	cout << "Wrote relation 2 to the disk" << endl;

	// Free the buffers
	DeAllocateVirtualBuffer((void**)(&m_psRel1Buffer), nBigRelByteCount);
	DeAllocateVirtualBuffer((void**)(&m_psRel2Buffer), nSmallRelByteCount);

	return true;

}

//// Description: in this sort function we are sorting in log(N) iterations. N is the number of sorted blocks - we are sorting in the
////					beginning and then we sort every 2^i blocks until we get 2^i=N
//template <class T>
//bool JoinTests::SortInIterations(int nCurrIteration, int nTotalNumOfIterations, wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
//											int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum, wstring& rwsFinalFileName)
//{
//
//	/////////////////////////////////////////////////////////////////////
//	/// Left TODO:	1. Handle the case of uneven buffers num (for example 5) - we have
//	///					the formula but we didn't implement it.
//	///				2. Fix a bug: every run we handle 2 sets of size nNumOfBlocksInIteration / 2 V
//	///				3. Check if it is working on small and big relations
//	///				4. Check what happen if the amount is smaller then sector (512)
//	/////////////////////////////////////////////////////////////////////
//
//	if (0 == nCurrIteration)
//	{
//		// This is the first iteration, we should sort block by block
//		CreateBlockSortedRelation<T>( wsRelFileName, wsSortedFileName, wsSortedFileName, nRelRowsNum,
//			nRelByteCount, rfTimeSec, rnReadSectorsNum, rnWrittenSectorsNum, false);
//		rwsFinalFileName = wsSortedFileName;
//	}
//	else
//	{
//		// Sort every 2^nCurrIteration blocks
//		int nNumOfBlocksInIteration		= static_cast<int>(pow(2.0f, nCurrIteration));
//		int nNumOfBlocksInLastIteration	= static_cast<int>(pow(2.0f, nCurrIteration-1));
//		int nEveryBufferByteCount		= static_cast<int>(nBufferLimitedByteCount / 2.0f);//static_cast<float>(nNumOfBlocksInIteration));
//		int nEveryBufferRowsNum			= (nEveryBufferByteCount / sizeof(T));
//		int nNumOfBlocks				=  static_cast<int>(ceil( static_cast<float>(nRelByteCount) / nBufferLimitedByteCount));
//		int nNumOfRuns					= static_cast<int>(ceil(nNumOfBlocks / static_cast<float>(nNumOfBlocksInIteration)));
//		int nNumOfLeftBlocks			= (nNumOfBlocks % nNumOfBlocksInIteration);
//		int nResBlockNumOfRows			= (nBufferLimitedByteCount/ sizeof(T));
//
//		T* psRelBuffer				= NULL;
//		T* psRelSoretdBuffer		= NULL;
//		// TODO: handle when nEveryBufferByteCount is less then a sector
//		if (nEveryBufferByteCount < 512 || 0 != (nEveryBufferByteCount % 512))
//		{
//			printf("SortInIterations - Problem - not 512 aligned\n");
//			return false;
//		}
//
//		if (AllocateVirtualBuffer((void**)(&psRelBuffer), nBufferLimitedByteCount ) == true)
//		{
//			if (AllocateVirtualBuffer((void**)(&psRelSoretdBuffer), nBufferLimitedByteCount ) != true)
//			{
//				DeAllocateVirtualBuffer((void**)(&psRelBuffer));
//				return false;
//			}
//		}
//
//		Disk* psDiskRel			= new Disk(wsSortedFileName, eLogical, eRead);
//		wsSortedFileName.append(L"1");
//		Disk* psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);
//
//		SSortedBufferInfo<T>* psBuffers	= (SSortedBufferInfo<T>*)malloc(2/*nNumOfBlocksInIteration*/ * sizeof(SSortedBufferInfo<T>));
//
//		if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
//		{
//			if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
//			{
//				DWORDLONG dwlOffset			= 0;
//				DWORDLONG dwlSoretdOffset	= 0;
//				int nSortedBufferIndex		= 0;
//
//				for (int i = 0; i < nNumOfRuns; ++i)
//				{
//					/*printf("Run %d \n", i);*/
//					int nFirstBufferSize	= nNumOfBlocksInLastIteration;
//					int nSecondBufferSize	= nNumOfBlocksInIteration;
//					
//					bool bLeftOver = false;
//					if ((nNumOfRuns - 1 == i) && (0  != nNumOfLeftBlocks))
//					{
//						// in the last iteration we have to merge the "leftovers"
//						/*printf("Must merge out of the regular merge\n");*/
//						
//						// We have to merge some blocks that we didn't merge in the regular 
//						// merge because their number was too small.
//						// for example: we have 7 blocks and we merged every 1, 2, 4, 8 blocks. 
//						// We didn't handle the last block when we merge every 2, but we have to
//						// merge the last 3 blocks when we merge every 4 blocks. Then when we merge
//						// every 8 blocks we will merge the first 4 blocks with the last 3 blocks.
//
//						if (nNumOfLeftBlocks < nNumOfBlocksInIteration && nNumOfLeftBlocks > nNumOfBlocksInLastIteration)
//						{
//							// We can merge this block with the blocks that were merge last iteration
//
//							nFirstBufferSize	= nNumOfBlocksInLastIteration;
//							nSecondBufferSize	= nNumOfLeftBlocks - nFirstBufferSize;
//
//							/*printf("merge block size %d with block size %d \n", nFirstBufferSize, nSecondBufferSize);*/
//							bLeftOver = true;
//
//						}
//						else
//						{
//							printf("Don't merge %d blocks in iteration %d with %d num of blocks in iteration \n",
//								nNumOfLeftBlocks, nCurrIteration, nNumOfBlocksInIteration);
//		
//							for (int j = 0; j < nNumOfLeftBlocks; ++j)
//							{
//								dwlOffset = (j+i*nNumOfBlocksInIteration) * nBufferLimitedByteCount;
//								if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
//								{
//									//rnReadSectorsNum += nBufferLimitedByteCount / 512;
//									if (false == psDiskSortedRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, dwlSoretdOffset, &rnWrittenSectorsNum))
//									{
//										printf("SortInIterations - Problem - failed reading\n");
//										break;
//									}
//									dwlSoretdOffset += nBufferLimitedByteCount;
//									
//									//rnWrittenSectorsNum += nBufferLimitedByteCount / 512;
//								}
//							}
//							break;
//						}
//					}
//					// Read for the first time - from evey block nEveryBufferByteCount
//					for (int j = 0; j < 2/*nNumOfBlocksInIteration*/; ++j)
//					{
//						dwlOffset	= ((j*nNumOfBlocksInLastIteration + i*nNumOfBlocksInIteration) * nBufferLimitedByteCount);
//						int nIndex	= (j*nEveryBufferRowsNum);
//						if (psDiskRel->DoOverlappedIO(psRelBuffer+nIndex, nEveryBufferByteCount, dwlOffset, &rnReadSectorsNum))
//						{
//							psBuffers[j].m_nTotalReadBytes = nEveryBufferByteCount / 512;
//							//rnReadSectorsNum += nEveryBufferByteCount / 512;
//							psBuffers[j].m_nLeftBytesToRead = (((nNumOfBlocksInIteration/2)*nBufferLimitedByteCount) -
//																	nEveryBufferByteCount);
//							psBuffers[j].m_pBuffer = psRelBuffer+nIndex;	
//							psBuffers[j].m_nCurrIndex = 0;
//							psBuffers[j].m_dwOffset = dwlOffset;	
//							psBuffers[j].m_bIsFinished = false;
//							if ((true == bLeftOver) && (1==j))
//							{
//								// This is the second buffer - its not whole
//								psBuffers[j].m_nLeftBytesToRead = ((nSecondBufferSize*nBufferLimitedByteCount) - 
//																	nEveryBufferByteCount);
//
//							}
//
//							/*printf("Read buffer %d - offset %d\n", j, dwlOffset);*/
//
//							//for (int k = 0; k < nEveryBufferByteCount/sizeof(T); ++k)
//							//{
//							//	printf("%d,", psBuffers[j].m_pBuffer[k]);
//							//}
//							//printf("\n");
//						}
//						else
//						{
//							printf("SortInIterations - Problem - failed reading\n");
//							break;
//						}
//					}
//					
//					bool bRes = false;
//					// Perform the merge
//					while(bRes == false)
//					{
//						// Get the smaller element of whole buffers
//						int nSmallerIndex = 0;
//						//for (int j = 0; j < 2; ++j)
//						//{
//						if (false == psBuffers[1].m_bIsFinished && false == psBuffers[nSmallerIndex].m_bIsFinished)
//						{
//							if (*(psBuffers[nSmallerIndex].m_pBuffer + psBuffers[nSmallerIndex].m_nCurrIndex) > 
//								*(psBuffers[1].m_pBuffer + psBuffers[1].m_nCurrIndex))
//							{
//								nSmallerIndex = 1;
//							}
//						}
//						else if (false == psBuffers[1].m_bIsFinished)
//						{
//							nSmallerIndex = 1;
//						}
//						//}
//						if (false == psBuffers[nSmallerIndex].m_bIsFinished)
//						{
//							// Write the smaller element to the result buffer
//							psRelSoretdBuffer[nSortedBufferIndex++] = 
//								*(psBuffers[nSmallerIndex].m_pBuffer + psBuffers[nSmallerIndex].m_nCurrIndex);
//
//							if(nSortedBufferIndex >= 2 && psRelSoretdBuffer[nSortedBufferIndex-1] <  psRelSoretdBuffer[nSortedBufferIndex-2])
//							{
//								printf("found - %d - %d\n", psRelSoretdBuffer[nSortedBufferIndex-1].GetKey(), psRelSoretdBuffer[nSortedBufferIndex-2].GetKey());
//
//							}
//							psBuffers[nSmallerIndex].m_nCurrIndex++;
//
//						}
//
//						// We have reached to the end of a reading buffer
//						if (psBuffers[nSmallerIndex].m_nCurrIndex >= nEveryBufferRowsNum && 
//							(false == psBuffers[nSmallerIndex].m_bIsFinished))
//						{
//							//if (nBufferLimitedByteCount == (psBuffers[nSmallerIndex].m_dwOffset - (nSmallerIndex + i) * nBufferLimitedByteCount))
//							//{
//							//	// There is no need to read more - this is the end of this block
//							//	psBuffers[nSmallerIndex].m_bIsFinished = true;
//							//}
//							//else 
//							if (0 == psBuffers[nSmallerIndex].m_nLeftBytesToRead)
//							{
//								// We have read the whole buffer
//								psBuffers[nSmallerIndex].m_bIsFinished = true;
//							}
//							else
//							{
//								// We have to read one more buffer
//								dwlOffset = psBuffers[nSmallerIndex].m_dwOffset + nEveryBufferByteCount;
//
//								int nReadByteCount = min(nEveryBufferByteCount, psBuffers[nSmallerIndex].m_nLeftBytesToRead);
//			
//								if (psDiskRel->DoOverlappedIO(psBuffers[nSmallerIndex].m_pBuffer, nReadByteCount, dwlOffset, &rnReadSectorsNum))
//								{
//									psBuffers[nSmallerIndex].m_nTotalReadBytes += nReadByteCount / 512;
//									//rnReadSectorsNum += nEveryBufferByteCount / 512;
//									psBuffers[nSmallerIndex].m_nCurrIndex = 0;
//									psBuffers[nSmallerIndex].m_dwOffset = dwlOffset;
//									psBuffers[nSmallerIndex].m_nLeftBytesToRead -= nReadByteCount;
//
//									/*printf("Read buffer %d - offset %d\n", nSmallerIndex, dwlOffset);*/
//									//for (int k = 0; k < nReadByteCount/sizeof(T); ++k)
//									//{
//									//	printf("%d,", psBuffers[nSmallerIndex].m_pBuffer[k]);
//									//}
//									//printf("\n");
//
//								}
//								else
//								{
//									printf("SortInIterations - Problem - failed reading\n");
//									break;
//								}
//							}
//
//						}
//
//						// We have reached to the end of the results buffer - we have to write the result to the disk
//						if (nResBlockNumOfRows <= nSortedBufferIndex)
//						{
//							if (false == psDiskSortedRel->DoOverlappedIO(psRelSoretdBuffer, nBufferLimitedByteCount, dwlSoretdOffset, &rnWrittenSectorsNum))
//							{
//								printf("SortInIterations - Problem - failed reading\n");
//								break;
//							}
//							/*printf("************Wrote to disk to offset %d \n", dwlSoretdOffset);*/
//							dwlSoretdOffset += nBufferLimitedByteCount;
//							nSortedBufferIndex = 0;
//							//rnWrittenSectorsNum += nBufferLimitedByteCount/512;
//						}
//
//						if ((true == psBuffers[0].m_bIsFinished) && (true == psBuffers[1].m_bIsFinished))
//						{
//							bRes = true;
//						}
//					}
//
//				}
//
//			}
//			psDiskRel->CloseDisk();
//			psDiskSortedRel->CloseDisk();
//		}
//
//		//rnReadSectorsNum += psBuffers[0].m_nTotalReadBytes;
//		//rnReadSectorsNum += psBuffers[1].m_nTotalReadBytes;
//		free(psBuffers);
//		delete(psDiskRel);
//		delete(psDiskSortedRel);
//		DeAllocateVirtualBuffer((void**)(&psRelBuffer));
//		DeAllocateVirtualBuffer((void**)(&psRelSoretdBuffer));
//
//				
//
//	}
//
//
//
//	if (nCurrIteration >= nTotalNumOfIterations-1)
//	{
//		// Finish
//		rwsFinalFileName = wsSortedFileName;
//		return true;
//	}
//	else
//	{
//		// Proceed with the iterations
//		SortInIterations<T>(nCurrIteration+1, nTotalNumOfIterations, wsRelFileName, wsSortedFileName, nRelRowsNum,
//			nRelByteCount, rfTimeSec, rnReadSectorsNum, rnWrittenSectorsNum, rwsFinalFileName);
//	}
//	return true;
//	
//}
//
//// Description: Suppose the memory size is M and the number of blocks of the relation is N. then every time we will sort together M/N 
////					elements from every block - in this way we are reading the relation only twice and write one intermediate results
////					and one final results.
//template <class T>
//bool JoinTests::SortInBlocks(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
//											int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
//{
//	////////////////////////////////////////////////////////////////////////////////////////////////
//	// For start we should sort block by block
//	CreateBlockSortedRelation<T>( wsRelFileName, wsSortedFileName, wsSortedFileName, nRelRowsNum,
//		nRelByteCount, rfTimeSec, rnReadSectorsNum, rnWrittenSectorsNum, false);
//
//	///////////////////////////////////////////////////////////////////////////////////////////////
//	// Start the merge of the sorted blocks to a one sorted file
//
//	int nBuffersLimitedRowsNum		= (nBufferLimitedByteCount / sizeof(T));
//	// number of blocks is the relation size / memory limit
//	int nNumOfBlocks				= static_cast<int>(ceil((static_cast<float>(nRelByteCount)) / nBufferLimitedByteCount));
//
//	// we will read from every sorted block M/N bytes
//	int nEveryBufferByteCount		= static_cast<int>(ceil(nBufferLimitedByteCount / static_cast<float>(nNumOfBlocks)));
//
//	// we will read from every sorted block M/N rows
//	int nEveryBufferRowsNum			= (nEveryBufferByteCount / sizeof(T));
//
//	// we will need x iterations in order to read the whole block
//	int nNumOfRuns					= static_cast<int>(ceil(nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));
//
//	// TODO: check if the case in which there is one last block which its size is smaller than others.
//
//	// we need N buffers in order to read them together
//	T** psRelBuffer				= (T**)malloc(nNumOfBlocks*sizeof(T*));
//	T* psRelSoretdBuffer		= NULL;
//
//	// TODO: handle when nEveryBufferByteCount is less then a sector
//	if (nEveryBufferByteCount < 512 || 0 != (nEveryBufferByteCount % 512))
//	{
//		printf("SortInIterations - Problem - not 512 aligned\n");
//		return false;
//	}
//
//	bool bRes = true;
//	int i = 0;
//
//	// Allocate virtual memory for the buffers
//	for (i = 0; i < nNumOfBlocks; ++i)
//	{
//		if (AllocateVirtualBuffer((void**)(&(psRelBuffer[i])), nEveryBufferByteCount ) != true)
//		{
//			bRes = false;
//			break;
//		}
//	}
//
//	if (AllocateVirtualBuffer((void**)(&psRelSoretdBuffer), nBufferLimitedByteCount ) != true)
//	{
//		bRes = false;		
//	}
//
//	if (false == bRes)
//	{
//		// Deallocate the previous buffers that were successfully been allocated
//		// TODO: check if we have deallocate until i or until i-1
//		for (int j = 0; j < i; ++j)
//		{
//			DeAllocateVirtualBuffer((void**)(&(psRelBuffer[j])));
//		}
//
//		return false;
//	}
//
//
//	// This disk is for reding the intermediate results of the sorted buffers.
//	Disk* psDiskRel			= new Disk(wsSortedFileName, eLogical, eRead);
//	
//	// The results file name will be wsSortedFileName + 1
//	wsSortedFileName.append(L"1");
//	Disk* psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);
//
//	// Allocating the meta-data structure for everyone of the buffers
//	SSortedBufferInfo<T>* psBuffers	= (SSortedBufferInfo<T>*)malloc(nNumOfBlocks * sizeof(SSortedBufferInfo<T>));
//
//	if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
//	{
//		if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
//		{
//			DWORDLONG dwlOffset			= 0;
//			DWORDLONG dwlSoretdOffset	= 0;
//			int nSortedBufferIndex		= 0;
//
//			// Read the first part of every buffer
//			for (int j = 0; j < nNumOfBlocks; ++j)
//			{
//				dwlOffset	= (j * nBufferLimitedByteCount);
//
//				int nIndex	= (j*nEveryBufferRowsNum);
//				
//				// Read the data
//				if (psDiskRel->DoOverlappedIO((psRelBuffer[j]), nEveryBufferByteCount, dwlOffset, &rnReadSectorsNum))
//				{
//					// Update the meta-data
//					psBuffers[j].m_nTotalReadBytes	= nEveryBufferByteCount;
//
//					//rnReadSectorsNum				+= nEveryBufferByteCount / 512;
//
//					psBuffers[j].m_nLeftBytesToRead	= (nBufferLimitedByteCount - nEveryBufferByteCount);
//
//					psBuffers[j].m_nCurrIndex = 0;
//
//					psBuffers[j].m_dwOffset = dwlOffset;	
//
//					// TODO: what if there is only one part to read
//					psBuffers[j].m_bIsFinished = false;
//
//					/*printf("Read buffer %d - offset %d\n", j, dwlOffset);*/
//
//					//for (int k = 0; k < nEveryBufferByteCount/sizeof(T); ++k)
//					//{
//					//	printf("%d,", psBuffers[j].m_pBuffer[k]);
//					//}
//					//printf("\n");
//				}
//				else
//				{
//					printf("SortInIterations - Problem - failed reading\n");
//					break;
//				}
//			}
//
//
//			// Perform the sorting
//			//for (int i = 0; i < nNumOfRuns; ++i)
//			//{
////					printf("Run %d \n", i);
//				
//			bool bRes = true;
//
//			while(bRes == true)
//			{
//				bRes = false;
//
//				T* sElem = NULL;
//
//				int nSmallerBufferIndex = 0;
//				
//				T* sMinElem = NULL;
//
//				int nFirstBuffer = 0;
//				// Get the first buffer that we didn't finish reading to be the first smaller one
//				for (int j = 0; j < nNumOfBlocks; ++j)
//				{
//					if (false == psBuffers[j].m_bIsFinished)
//					{
//						// The first buffere which we didn't finish reading
//						sMinElem = (psRelBuffer[j]) + psBuffers[j].m_nCurrIndex;
//						nFirstBuffer = j;
//						nSmallerBufferIndex = j;
//						bRes = true;
//						break;
//					}
//
//				}				
//
//				// Get the smaller element of whole buffers
//				for (int j = nFirstBuffer+1; j < nNumOfBlocks; ++j)
//				{
//					if (false == psBuffers[j].m_bIsFinished)
//					{
//						sElem = (psRelBuffer[j]) + psBuffers[j].m_nCurrIndex;
//						
//						if (*sElem < *sMinElem)
//						{
//							// We have smaller element
//							nSmallerBufferIndex = j;
//							sMinElem = sElem;
//						}
//
//						// We still have buffers that we didn't finish reading
//						bRes = true;
//					}
//				}
//
//				if (true == bRes)
//				{
//					// Write the smaller element to the results buffer
//					psRelSoretdBuffer[nSortedBufferIndex++] = *(sMinElem);
//
//					// Increase the current index
//					psBuffers[nSmallerBufferIndex].m_nCurrIndex++;
//
//					// We have reached to the end of this part - must read another part
//					if (nEveryBufferRowsNum <= psBuffers[nSmallerBufferIndex].m_nCurrIndex)
//					{
//						if (0 == psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead)
//						{
//							// We have read the whole buffer
//							psBuffers[nSmallerBufferIndex].m_bIsFinished = true;
//						}
//						else
//						{
//							// We have to read one more part
//							dwlOffset = psBuffers[nSmallerBufferIndex].m_dwOffset + nEveryBufferByteCount;
//
//							// Maybe the left bytes to read is smaller than the part size
//							int nReadByteCount = min(nEveryBufferByteCount, psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead);
//		
//							if (psDiskRel->DoOverlappedIO((psRelBuffer[nSmallerBufferIndex]), nReadByteCount, dwlOffset, &rnReadSectorsNum))
//							{
//								psBuffers[nSmallerBufferIndex].m_nTotalReadBytes	+= nReadByteCount;
//								//rnReadSectorsNum									+= nReadByteCount / 512;
//								psBuffers[nSmallerBufferIndex].m_nCurrIndex			= 0;
//								psBuffers[nSmallerBufferIndex].m_dwOffset			= dwlOffset;
//								psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead	-= nReadByteCount;
//								psBuffers[nSmallerBufferIndex].m_bIsFinished		= false;
//
//								/*printf("Read buffer %d - offset %d\n", nSmallerIndex, dwlOffset);*/
//								//for (int k = 0; k < nReadByteCount/sizeof(T); ++k)
//								//{
//								//	printf("%d,", psBuffers[nSmallerIndex].m_pBuffer[k]);
//								//}
//								//printf("\n");
//
//							}
//							else
//							{
//								printf("SortInIterations - Problem - failed reading\n");
//								break;
//							}
//						}
//					}
//				}
//				else
//				{
//					printf("bla\n");
//				}
//
//
//				// If we have reached to the end of the results buffer or we finished reading all blocks 
//				//		we have to write the results to the disk
//				if ((nBuffersLimitedRowsNum <= nSortedBufferIndex || false == bRes) && (nSortedBufferIndex > 0))
//				{
//					if (false == psDiskSortedRel->DoOverlappedIO(psRelSoretdBuffer, nSortedBufferIndex*sizeof(T), dwlSoretdOffset, &rnWrittenSectorsNum))
//					{
//						printf("SortInIterations - Problem - failed reading\n");
//						break;
//					}
//					/*printf("************Wrote to disk to offset %d \n", dwlSoretdOffset);*/
//					dwlSoretdOffset += nBufferLimitedByteCount;
//					//rnWrittenSectorsNum += nBufferLimitedByteCount/512;
//					nSortedBufferIndex = 0;
//				}
//
//			}
//
//		}
//
//	}
//	psDiskRel->CloseDisk();
//	psDiskSortedRel->CloseDisk();
//	//}
//
//	free(psBuffers);
//	delete(psDiskRel);
//	delete(psDiskSortedRel);
//	
//	// Deallocate the memory for the read-buffers
//	for (int j = 0; j < nNumOfBlocks; ++j)
//	{
//		DeAllocateVirtualBuffer((void**)(&(psRelBuffer[j])));
//	}
//
//	// Deallocate the memory for the result buffer
//	DeAllocateVirtualBuffer((void**)(&psRelSoretdBuffer));	
//
//	return bRes;
//
//}
//
//
//// Description: Read the whole relation and keep the k smallest elements. Then write them and read the relation again and keep the 
////	next k smallest elements, and so on.
////	We will write the relation once and read the relation logk(M) times.
//
//template <class T>
//bool JoinTests::MergeSortKsV1(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
//					int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
//{
//	rnReadSectorsNum		= 0;
//	rnWrittenSectorsNum		= 0;
//
//	bool bRes = true;
//
//	// Initialization
//	m_nMaxElemInList	= 0;
//	// we keep the multiplicity of the maximum number
//	m_nMaxElemMul		= 1;
//
//	// the results buffer and the read buffer are the same size.
//	int nBuffersLimitedRowsNum		= (nBufferLimitedByteCount / sizeof(T));
//
//	int nNumberOfBuffersToRead		= static_cast<int>(ceil(nRelByteCount /  static_cast<float>(nBufferLimitedByteCount)));
//
//	unsigned int  nMaxResultSize	= nBuffersLimitedRowsNum;
//	// We will need log(M) iterations - we will read the whole relation and take the smallest M.
//	int nNumOfRuns					= nNumberOfBuffersToRead;//static_cast<int>(ceil(nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));
//
//	// Allocate buffers
//
//	T* psRelBuffer				= NULL;
//	T* psRelResultBuffer		= NULL;
//	T* psResultBuffer			= NULL;
//
//	if (AllocateVirtualBuffer((void**)(&psRelBuffer), nBufferLimitedByteCount ) == true)
//	{
//		// Handle two results buffer 
//		if (AllocateVirtualBuffer((void**)(&psResultBuffer), 2*nBufferLimitedByteCount ) != true)
//		{
//			DeAllocateVirtualBuffer((void**)(&psRelBuffer));
//			return false;
//		}
//	}
//
//	psRelResultBuffer	= psResultBuffer;
//	int nResultOffset	= 0;
//	int nResIndex		= 0;
//
//	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
//	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);
//
//	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
//	{
//		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
//		{
//			DWORDLONG dwlOffset			= 0;
//			DWORDLONG dwlResultOffset	= 0;
//
//			m_nMaxElemInList	= -1;
//			m_nMaxElemMul		= 0;
//
//			for (int j = 0; j < nNumOfRuns; ++j)
//			{
//				psRelResultBuffer	= psResultBuffer;
//				nResultOffset		= nBuffersLimitedRowsNum;
//
//				// 1. Read the first buffer of the relation
//
//				if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
//				{		
//					dwlOffset+= nBufferLimitedByteCount;
//
//					// 1.1 Sort the first buffer of the relation
//					QuickSort(psRelBuffer, 0, nBuffersLimitedRowsNum);
//
//					// 1.3 Copy it to the result buffer
//					memcpy(psRelResultBuffer, psRelBuffer, nBufferLimitedByteCount);
//
//					// 2. For the next buffers of the relation
//
//					for (int i = 0; i < nNumberOfBuffersToRead - 1; ++i)
//					{
//						// 2.1 Read the buffer
//						if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
//						{		
//							dwlOffset+= nBufferLimitedByteCount;
//
//							// 2.2 Sort the buffer
//							QuickSort(psRelBuffer, 0, nBuffersLimitedRowsNum);
//
//							if (psRelBuffer[0] <= psRelResultBuffer[nBuffersLimitedRowsNum-1])
//							{
//								// 2.3 Merge the new buffer with the last one and take the smallest M
//								
//								int nIndex1 = 0;
//								int nIndex2 = 0;
//								nResIndex	= 0;
//
//
//								// Perform a merge sort between the new buffer and the result buffer
//								// In order not to copy a lot of elements we will have another buffer of pointers and
//								// in the end we will copy them to the result buffer.
//
//								T** aBuffer = (T**)malloc(nBuffersLimitedRowsNum*sizeof(T*));
//
//								int nCurrElemMul = 0;
//
//								// TODO: handle the case in which the first buffer was not full because
//								//			there were not enough elements bigger then the last maximum
//
//								// 2.3.1 Iterate over the two buffers
//								while (	nIndex1 < nBuffersLimitedRowsNum && 
//										nIndex2 < nBuffersLimitedRowsNum && 
//										nResIndex < nBuffersLimitedRowsNum )
//								{
//									if ((m_nMaxElemInList > psRelBuffer[nIndex1].GetKey()))
//									{
//										// the elemets are smaller then the last maximum number
//										// they already been handle
//										nIndex1++;
//									}
//									else if ((m_nMaxElemInList > psRelResultBuffer[nIndex2].GetKey()))
//									{
//										// the elemets are smaller then the last maximum number
//										// they already been handle
//										nIndex2++;
//									}
//									else if (m_nMaxElemInList == psRelBuffer[nIndex1].GetKey() && nCurrElemMul < m_nMaxElemMul)
//									{
//										nCurrElemMul++;
//										nIndex1++;
//									}
//									else if (m_nMaxElemInList == psRelResultBuffer[nIndex2].GetKey() && nCurrElemMul < m_nMaxElemMul)
//									{
//										nCurrElemMul++;
//										nIndex2++;
//									}
//									else
//									{
//										if (psRelBuffer[nIndex1] < psRelResultBuffer[nIndex2])
//										{
//											aBuffer[nResIndex] = &(psRelBuffer[nIndex1]);
//											++nResIndex;
//											++nIndex1;
//										}
//										else
//										{
//											aBuffer[nResIndex] = &(psRelResultBuffer[nIndex2]);
//											++nResIndex;
//											++nIndex2;
//										}
//									}
//
//								}
//
//								// 2.3.2 Copy the rest of the list (if one of the lists were over)
//								if (nResIndex < nBuffersLimitedRowsNum)
//								{
//									if (nIndex1 < nBuffersLimitedRowsNum)
//									{
//										while (nResIndex < nBuffersLimitedRowsNum && nIndex1 < nBuffersLimitedRowsNum)
//										{
//											aBuffer[nResIndex] = &(psRelBuffer[nIndex1]);
//											++nResIndex;
//											++nIndex1;
//										}
//									}
//									else
//									{
//										while (nResIndex < nBuffersLimitedRowsNum && nIndex2 < nBuffersLimitedRowsNum)
//										{
//											aBuffer[nResIndex] = &(psRelResultBuffer[nIndex2]);
//											++nResIndex;
//											++nIndex2;
//										}
//									}
//								}
//
//								// 2.3.3 Switch the result buffer pointer
//								psRelResultBuffer = psResultBuffer + nResultOffset;
//
//								if (0 == nResultOffset)
//								{
//									nResultOffset = nBuffersLimitedRowsNum;
//								}
//								else
//								{
//									nResultOffset = 0;
//								}
//								
//								// 2.3.4 Copy the pointers into the results buffer
//
//								for (nIndex1 = 0; nIndex1 < nResIndex;  ++nIndex1)
//								{
//									psRelResultBuffer[nIndex1] = *(aBuffer[nIndex1]);
//								}
//							}
//
//						}
//						else
//						{
//							// the sorted new buffer first element is already bigger then the first
//							// element of the result buffer - no merge is needed.
//						}
//					}
//
//				}
//				
//				// 4.1 Get the maximum number of this buffer and its multiplicity
//				
//				m_nMaxElemInList	= psRelResultBuffer[nBuffersLimitedRowsNum-1].GetKey();
//
//				for (int i = nBuffersLimitedRowsNum-1; i > 0; --i)
//				{
//					if (psRelResultBuffer[i].GetKey() == m_nMaxElemInList)
//					{
//						m_nMaxElemMul++;
//					}
//					else
//					{
//						break;
//					}
//				}
//
//				// 4.2 Write the result list 
//				if (true == psDiskResult->DoOverlappedIO(psRelResultBuffer, nBufferLimitedByteCount, dwlResultOffset, &rnWrittenSectorsNum))
//				{							
//					dwlResultOffset += nBufferLimitedByteCount;
//				}
//
//			}
//		}
//
//	}
//
//	psDiskRel->CloseDisk();
//	psDiskResult->CloseDisk();
//
//	delete(psDiskRel);
//	delete(psDiskResult);
//
//	DeAllocateVirtualBuffer((void**)(&psRelBuffer));
//
//	DeAllocateVirtualBuffer((void**)(&psResultBuffer));
//
//	return bRes;
//
//
//
//}
//
//
//template <class T>
//bool JoinTests::MergeSortKsV2(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
//					int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
//{
//	rnReadSectorsNum		= 0;
//	rnWrittenSectorsNum		= 0;
//
//	bool bRes = true;
//
//	// Initialization
//	m_nMaxElemInList	= 0;
//	// we keep the multiplicity of the maximum number
//	m_nMaxElemMul		= 1;
//
//	// the results buffer and the read buffer are the same size.
//	int nBuffersLimitedRowsNum		= (nBufferLimitedByteCount / sizeof(T));
//
//	int nNumberOfBuffersToRead		= static_cast<int>(ceil(nRelByteCount /  static_cast<float>(nBufferLimitedByteCount)));
//
//	unsigned int  nMaxResultSize	= nBuffersLimitedRowsNum;
//	// We will need log(M) iterations - we will read the whole relation and take the smallest M.
//	int nNumOfRuns					= nNumberOfBuffersToRead;//static_cast<int>(ceil(nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));
//
//	// Allocate buffers
//
//	T* psRelBuffer				= NULL;
//	T* psResultBuffer			= NULL;
//
//	if (AllocateVirtualBuffer((void**)(&psResultBuffer), 2*nBufferLimitedByteCount ) == false)
//	{
//		return false;
//	}
//
//	psRelBuffer	= psResultBuffer;
//
//	int nResIndex		= 0;
//	int nIndex			= 0;
//	int nCurrMaxMul		= 0;
//	int nLastItrElems	= 0;
//
//	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
//	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);
//
//	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
//	{
//		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
//		{
//			DWORDLONG dwlOffset			= 0;
//			DWORDLONG dwlResultOffset	= 0;
//
//			m_nMaxElemInList	= -1;
//			m_nMaxElemMul		= 0;
//
//			for (int j = 0; j < nNumOfRuns; ++j)
//			{			
//				int nBufferOffset	= 0;
//				int nNumOfElements	= 0;
//				nCurrMaxMul			= 0;
//				nLastItrElems		= 0;
//				dwlOffset			= 0;
//
//				for (int i = 0; i < nNumberOfBuffersToRead; ++i)
//				{
//					psRelBuffer	= psResultBuffer + nBufferOffset;
//							
//					// 1. Read the first buffer of the relation
//
//					if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
//					{		
//						dwlOffset+= nBufferLimitedByteCount;
//
//						nResIndex	= 0;
//						nIndex		= 0;
//
//						// Take only the relevant elements.
//						for (nIndex = 0; nIndex < nBuffersLimitedRowsNum; ++nIndex)
//						{
//							if (psRelBuffer[nIndex].GetKey() < m_nMaxElemInList)
//							{
//								continue;
//							}
//
//							if (psRelBuffer[nIndex].GetKey() == m_nMaxElemInList && nCurrMaxMul < m_nMaxElemMul)
//							{
//								nCurrMaxMul++;
//								continue;
//							}
//							
//							psRelBuffer[nResIndex] = psRelBuffer[nIndex];
//							++nResIndex;
//
//
//						}
//
//						nBufferOffset = nBuffersLimitedRowsNum;
//
//					}
//
//					if (0 == i)
//					{
//						nNumOfElements	= nResIndex;
//					}
//					else
//					{
//						nNumOfElements = min(nLastItrElems + nResIndex, 2*nBuffersLimitedRowsNum);
//
//					}
//
//					if (nResIndex < 640)
//					{
////						printf("nResIndex < 640\n");
//					}
////					nNumOfElements = min(nBuffersLimitedRowsNum + nResIndex, 2*nBuffersLimitedRowsNum);
//
//					// 1.1 Sort the buffer of the relation
//					QuickSort(psResultBuffer, 0, nNumOfElements);
//
//					nLastItrElems = nNumOfElements;
//
//					psRelBuffer	= psResultBuffer + min(nBufferOffset, nResIndex);
//
//				}
//
//				// 4.1 Get the maximum number of this buffer and its multiplicity
//				
//				m_nMaxElemInList	= psResultBuffer[nBuffersLimitedRowsNum-1].GetKey();
//
//				for (int i = nNumOfElements-1; i > 0; --i)
//				{
//					if (psResultBuffer[i].GetKey() == m_nMaxElemInList)
//					{
//						m_nMaxElemMul++;
//					}
//					else
//					{
//						break;
//					}
//				}
//
//				// 4.2 Write the result list 
//				if (true == psDiskResult->DoOverlappedIO(psResultBuffer, nBufferLimitedByteCount, dwlResultOffset, &rnWrittenSectorsNum))
//				{							
//					dwlResultOffset += nBufferLimitedByteCount;
//				}
//			}
//		}
//	}
//
//
//	psDiskRel->CloseDisk();
//	psDiskResult->CloseDisk();
//
//	delete(psDiskRel);
//	delete(psDiskResult);
//
//	DeAllocateVirtualBuffer((void**)(&psResultBuffer));
//
//	return bRes;
//}
//// Description: Read the whole relation and keep the k smallest elements. Then write them and read the relation again and keep the 
////	next k smallest elements, and so on.
////	We will write the relation once and read the relation logk(M) times.
//
//template <class T>
//bool JoinTests::MergeSortKs(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
//						int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
//{
//	rnReadSectorsNum		= 0;
//	rnWrittenSectorsNum		= 0;
//
//	bool bRes = true;
//
//	// Initialization
//	m_nMaxElemInList	= 0;
//	// we keep the multiplicity of the maximum number
//	m_nMaxElemMul		= 1;
//
//	// the results buffer and the read buffer are the same size.
//	int nBuffersLimitedRowsNum		= (nBufferLimitedByteCount / sizeof(T));
//
//	int nNumberOfBuffersToRead		= static_cast<int>(ceil(nRelByteCount /  static_cast<float>(nBufferLimitedByteCount)));
//
//	unsigned int  nMaxResultSize	= nBuffersLimitedRowsNum;
//	// We will need log(M) iterations - we will read the whole relation and take the smallest M.
//	int nNumOfRuns					= nNumberOfBuffersToRead;//static_cast<int>(ceil(nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));
//
//	// Allocate buffers
//
//	T* psRelBuffer				= NULL;
//	T* psRelResultBuffer		= NULL;
//
//	if (AllocateVirtualBuffer((void**)(&psRelBuffer), nBufferLimitedByteCount ) == true)
//	{
//		if (AllocateVirtualBuffer((void**)(&psRelResultBuffer), nBufferLimitedByteCount ) != true)
//		{
//			DeAllocateVirtualBuffer((void**)(&psRelBuffer));
//			return false;
//		}
//	}
//
//	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
//	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);
//
//	list<T> pSmallersList;
//
//	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
//	{
//		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
//		{
//			DWORDLONG dwlOffset			= 0;
//			DWORDLONG dwlResultOffset	= 0;
//
//			// Run over all iterations
//				
//			for (int i = 0; i < nNumOfRuns; ++ i)
//			{
//				dwlOffset		= 0;
//				int nLastMax	= m_nMaxElemInList;
//				int nLastMaxMul	= m_nMaxElemMul;
//				int nFoundMax	= 0;
//
//				// For the whole relation			
//				for (int j = 0; j < nNumberOfBuffersToRead; ++j)
//				{
//					// TODO: handle reading a last not whole buffer
//					// Read M bytes from the relation
//					if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
//					{		
//						//rnReadSectorsNum+= nBufferLimitedByteCount/512;
//						dwlOffset+= nBufferLimitedByteCount;
//						// Update the smallest M list
//						for (int k = 0; k < nBuffersLimitedRowsNum; ++k)
//						{
//							// Only if the element is bigger then the last maximum - inser it
//							if ((nLastMax < psRelBuffer[k].GetKey()) || 
//								((nLastMax == psRelBuffer[k].GetKey()) && (nFoundMax++ >= nLastMaxMul)))
//							{
//								
//								InsertToResultList<T>(&(psRelBuffer[k]), &pSmallersList);
//							}
//						}
//					}
//					else
//					{
//						printf("Failed reading the relation \n");
//					}
//				}
//				
//				// Copy the data from the list to the buffer
//
//				list<T>::iterator itr;
//				int nIndex = 0;
//
//				for(itr=pSmallersList.begin(); itr != pSmallersList.end(); ++itr)
//				{
//					psRelResultBuffer[nIndex] = *itr;
//					++nIndex;
//				}
//
//				// Clear the list
//				pSmallersList.clear();
//
//				// Write the result list 
//				if (true == psDiskResult->DoOverlappedIO(psRelResultBuffer, nBufferLimitedByteCount, dwlResultOffset, &rnWrittenSectorsNum))
//				{							
//					//rnWrittenSectorsNum+= nBufferLimitedByteCount/512;
//
//					dwlResultOffset += nBufferLimitedByteCount;
//				}
//		
//			}
//
//		}
//	}
//
//	psDiskRel->CloseDisk();
//	psDiskResult->CloseDisk();
//
//	delete(psDiskRel);
//	delete(psDiskResult);
//
//	DeAllocateVirtualBuffer((void**)(&psRelBuffer));
//
//	DeAllocateVirtualBuffer((void**)(&psRelResultBuffer));
//
//
//
//
//
//	// Run over all iterations
//		// For the whole relation
//			// Read the M bytes from the relation
//				// Update the smallest M list
//		// Write the list. 
//		// Keep the bigger number of this list.
//			// TODO: handle the case in which the bigger number in the list appears more than once, but not the whole copies are in
//			//			the written list - so the next time we will need to write the other copies.
//
//
//	return bRes;
//}
//
//template <class T>
//bool JoinTests::InsertToResultList(T* pNewElem, list<T>* pSmallersList)
//{
//	bool bRes = false;
//
//	// the results buffer and the read buffer are the same size.
//	int nBuffersLimitedRowsNum		= (nBufferLimitedByteCount / sizeof(T));
//
//	unsigned int  uMaxResultSize	= nBuffersLimitedRowsNum;
//
//	if (pNewElem->GetKey() > m_nMaxElemInList)
//	{
//		if (pSmallersList->size() < uMaxResultSize)
//		{
//			// If we have room in the list - 
//			// Insret it to the end of the list and declare a new maximum
//			pSmallersList->push_back(*pNewElem);
//			m_nMaxElemInList = pNewElem->GetKey();
//			// we keep the multiplicity of the maximum number
//			m_nMaxElemMul  = 1;
//			bRes = true;
//		}
//		else
//		{
//			// If the list is full - 
//			// don't enter the list
//		}
//	}
//	else if (pNewElem->GetKey() == m_nMaxElemInList)
//	{
//		// The new element key is as the last element in the list (maximum)
//		if (pSmallersList->size() < uMaxResultSize)
//		{
//			// Inset to the end of the list and declare a new maximum
//			pSmallersList->push_back(*pNewElem);
//			
//			// Increment the multiplicity of the maximum key in the list
//			m_nMaxElemMul++;
//
//			bRes = true;
//		}
//		else
//		{
//			// Don't enter the list
//		}
//
//	}
//	else
//	{
//		// the new element has to be insert in the middle of the list
//		list<T>::iterator itr;
//		list<T>::iterator itr1;
//		itr1 = (pSmallersList->end())--;
//
//		for(itr=pSmallersList->begin(); itr != pSmallersList->end(); ++itr)
//		{
//			// Go over the whole list and find its place
//			// TODO: maybe change it to a binary search
//			if (*pNewElem <= *itr)
//			{
//				if (pSmallersList->size() < uMaxResultSize)
//				{
//					// The list is not full
//					pSmallersList->insert(itr, *pNewElem);
//					bRes = true;
//				}
//				else
//				{
//					// The list is full
//
//					// Remove the last element and insert the new one in the proper place
//					itr = pSmallersList->insert(itr,*pNewElem);
//					//itr1 = pSmallersList->erase(itr1);
//					pSmallersList->pop_back();
//
//					bRes = true;
//
//					if (m_nMaxElemMul > 1)
//					{
//						// The maximum number is still the same
//						// decrease the multiplicity
//						m_nMaxElemMul--;
//
//					}
//					else
//					{
//					
//						// We have to find a new maximum
//						// Check what is the multiplicity of the new max element
//						itr1 = --pSmallersList->end();
//						m_nMaxElemInList = (*itr1).GetKey();
//						m_nMaxElemMul = 0;
//
//						// Go over the list (in reverse) and calc the multiplicity
//						while((*itr1).GetKey() == m_nMaxElemInList)
//						{
//							--itr1;
//							m_nMaxElemMul++; 
//						}
//					}
//				}
//				// we inserted the new element - we can stop
//				break;
//
//			}
//
//		}
//	}
//
//	return bRes;
//}

bool JoinTests::AllocateVirtualBuffer(void** pBuffer, int nBytesCount)
{
	bool bIsReady = false;
	// Allocate a buffer
	VirtualFree(*pBuffer, 0, MEM_RELEASE);
	if (!(*pBuffer = VirtualAlloc(NULL, nBytesCount,  MEM_COMMIT, PAGE_READWRITE/*|PAGE_NOCACHE*/)))
	{
		cout << "*** Could not allocate buffer to prepare disk." << endl;
		bIsReady =  false;
	}
	else
	{
		VirtualLock((void**)(pBuffer), nBytesCount);
		bIsReady = true;
	}
	return bIsReady;
}

bool JoinTests::DeAllocateVirtualBuffer(void** pBuffer, int nBytesCount)
{
	VirtualFree(*pBuffer, 0, MEM_RELEASE);
	VirtualUnlock((void**)(pBuffer), nBytesCount);
	return true;
}


BOOL JoinTests::GetLogicalDiskPhysicalInfo(wstring wDiskName)
{
	BOOL bResult = FALSE;

	bResult = GetDiskFreeSpace (wDiskName.c_str(), 
                                 &m_dwSectPerClust,
                                 &m_dwBytesPerSect, 
                                 &m_dwFreeClusters,
                                 &m_dwTotalClusters);

	if (!bResult)
	{
		cout <<"error: "<<  GetLastError(); 
		std::wcout << "could not get free space for "<< wDiskName << endl;
	}
	else
	{
		std::wcout << "Device name: \t\t\t" << wDiskName << endl;
		cout <<"Bytes per sectotr: \t\t" << m_dwBytesPerSect << endl;
		cout <<"Sectors per clusters: \t\t" << m_dwSectPerClust << endl;
		cout << "Total number of clusters: \t" <<  m_dwTotalClusters << endl;
		cout << "Number of free clusters : \t" << m_dwFreeClusters << endl;
		
		cout << endl;
		double fTotalSize = (m_dwBytesPerSect * m_dwSectPerClust) * 
			static_cast<float>((m_dwTotalClusters) / pow(10.0, 9.0));
		cout << "Calculate device size: \t\t "<< fTotalSize <<" GB " <<endl;
		fTotalSize = (m_dwBytesPerSect * m_dwSectPerClust) * static_cast<float>((m_dwFreeClusters) / pow(10.0, 9.0));
		cout << "Calculate device size (free): \t "<< fTotalSize <<" GB " <<endl;

	}

	return bResult;

}

void JoinTests::PreTest(wstring wsRel1FileNameFlash, wstring wsRel2FileNameFlash, wstring wsResFileNameFlash/*,
										wstring wsRel1FileNameDisk, wstring wsRel2FileNameDisk*/)
{
	m_wsRel1FileNameFlash	= wsRel1FileNameFlash;
	m_wsRel2FileNameFlash	= wsRel2FileNameFlash;
	m_wsResFileNameFlash	= wsResFileNameFlash;
	//m_wsRel1FileNameDisk	= wsRel1FileNameDisk;
	//m_wsRel2FileNameDisk	= wsRel2FileNameDisk;

	m_nRel1ByteCount = nBigRelByteCount;
	m_nRel2ByteCount = nSmallRelByteCount;

	m_nRel1RowsNum = m_nRel1ByteCount / REL_ROW_SIZE_BYTES;
	m_nRel2RowsNum = m_nRel2ByteCount  / REL_ROW_SIZE_BYTES ;

}

void JoinTests::PrintResults(ostream& fResults, float fTimeSec, int nReadSectorsNum)
{
	fResults << fTimeSec / 1000.0f<< "\t" ;/*<< "-";*/
	fResults << nReadSectorsNum << "\t\t";
}

void JoinTests::Test2(ostream& fResults, int nIteration)
{
	if (0 == nIteration)
	{
		fResults << "Memory Size[B] \t | Rel2 x Rel1 \t | Matches Num \t| Nested \t |" << " Offset-Hash \t |" << "Sorted-Block out\t |" << "Sorted-Block in\t |" <<
		"Sorted-Block RS NoSort\t |" <<"Sorted-Block RS out\t |" << "Sorted-Block RS in\t |" <<"Sorted-Local-Block \t | " << "Sorted-Local-Block In\t | " << endl;
		fResults << "------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------" << endl;
	}

	int nMatchesNum = 0;
	int nNumberOfWords = 0;
#ifdef __FLASH__
	wstring wsOffsetHashFile = L"\\\\.\\E:\\Hash";
#else
	wstring wsOffsetHashFile = L"\\\\.\\C:\\Hash";
#endif //	__FLASH__
	int nReadSectorsNum = 705536/*0*/;
		
	fResults << nBufferLimitedByteCount << "\t\t";
	fResults << nSmallRelByteCount << "X" << nBigRelByteCount << "\t\t";
	//// Create the relations
	//CreateRandomRelations(0.1f, 1.0f, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
	//						m_wsRel1FileNameFlash, m_wsRel2FileNameFlash/*, 
	//						m_wsRel1FileNameDisk, m_wsRel2FileNameDisk*/);

	float fTimeSec = 0;
	
	cout << endl << "Creating Hash to REL1" << endl;
	cout << "***************************************************** " << endl;

	fTimeSec = 0;
	float fHashTime = 0.516f/*0*/;

	//// Creating the offset-hash file
	//CreateOffsetHash( m_wsRel1FileNameFlash, wsOffsetHashFile, 0.5f, nNumberOfWords, fHashTime);

	nNumberOfWords = 4096;
	
	cout << endl << "Performing nested-loop join " << endl;
	cout << "***************************************************** " << endl;
	

	NestedLoopJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, m_wsResFileNameFlash, fTimeSec, nMatchesNum, nReadSectorsNum);
	
	fResults << nMatchesNum << "\t\t";
	PrintResults(fResults, fTimeSec, nReadSectorsNum);


//	cout << endl << "Performing sorted block join (No Rs  + out) " << endl;
//	cout << "***************************************************** " << endl;
//	
//	fTimeSec = 0;
//	fHashTime = 0;
//#ifdef __FLASH__
//	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
//	L"\\\\.\\E:\\Rel2Sorted", L"\\\\.\\E:\\rel2RS", false, 0.5f, fTimeSec, nNumberOfWords, nReadSectorsNum, eOuterSort);
//#else
//	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
//	L"\\\\.\\C:\\Rel2Sorted", L"\\\\.\\C:\\rel2RS", false, 0.1f, fTimeSec, nNumberOfWords, nReadSectorsNum, eOuterSort);
//#endif // __FLASH__
//
//	PrintResults(fResults, fTimeSec + fHashTime, nReadSectorsNum);


	cout << endl << "Performing local sorted block join with inner sort" << endl;
//	cout << endl << "Performing local sorted block join with no sort" << endl;
	cout << "***************************************************** " << endl;
	fTimeSec = 0;
#ifdef __FLASH__
	SortedBlockLocalJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, L"\\\\.\\E:\\Hash1",
		L"\\\\.\\E:\\SortRel1",  L"\\\\.\\E:\\rel1RS", false, 0.5f, fTimeSec, nReadSectorsNum, eNoSort);
#else
	SortedBlockLocalJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, L"\\\\.\\C:\\Hash1",
		L"\\\\.\\C:\\SortRel1",  L"\\\\.\\C:\\rel1RS", false, 0.1f, fTimeSec, nReadSectorsNum, eInnerSort);
#endif // __FLASH__
	PrintResults(fResults, fTimeSec, nReadSectorsNum);
	
}
void JoinTests::PerformTest(ostream& fResults, int nIteration)
{
	//CreateAlmostSortedRelation(0.1f, m_nRel2RowsNum, m_nRel2ByteCount, L"\\\\.\\C:\\TempFile");
	//SortingUnitTest();
	
	Test3(fResults, nIteration);
}
void JoinTests::Test1(ostream& fResults, int nIteration)
{
	if (0 == nIteration)
	{
		fResults << "Memory Size[B] \t | Rel2 x Rel1 \t | Matches Num \t| Nested \t |" << " Offset-Hash \t |" << "Sorted-Block out\t |" << "Sorted-Block in\t |" <<
		"Sorted-Block RS NoSort\t |" <<"Sorted-Block RS out\t |" << "Sorted-Block RS in\t |" <<"Sorted-Local-Block \t | " << "Sorted-Local-Block In\t | " << endl;
		fResults << "------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------" << endl;
	}

	int nMatchesNum = 0;
	int nNumberOfWords = 0;
#ifdef __FLASH__
	wstring wsOffsetHashFile = L"\\\\.\\E:\\Hash";
#else
	wstring wsOffsetHashFile = L"\\\\.\\C:\\Hash";
#endif //	__FLASH__
	int nReadSectorsNum = 0;
		
	fResults << nBufferLimitedByteCount << "\t\t";
	fResults << nSmallRelByteCount << "X" << nBigRelByteCount << "\t\t";
	//// Create the relations
	CreateRandomRelations(0.1f, 1.0f, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
							m_wsRel1FileNameFlash, m_wsRel2FileNameFlash/*, 
							m_wsRel1FileNameDisk, m_wsRel2FileNameDisk*/);

////#ifdef __FLASH__
////	CreateRandomRelations1(0.5f, 1, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
////							m_wsRel1FileNameFlash, m_wsRel2FileNameFlash);
////
////#else
////	CreateRandomRelations1(0.2f, 1, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
////							m_wsRel1FileNameDisk, m_wsRel2FileNameDisk);
//
////#endif // __FLASH__
//
//

	float fTimeSec = 0;

	if (m_nRel1ByteCount != 10000*1024)
	{
		cout << endl << "Performing nested-loop join " << endl;
		cout << "***************************************************** " << endl;
		

		NestedLoopJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, m_wsResFileNameFlash, fTimeSec, nMatchesNum, nReadSectorsNum);
		
		fResults << nMatchesNum << "\t\t";
		PrintResults(fResults, fTimeSec, nReadSectorsNum);
	}
	
	cout << endl << "Performing offset-hash join " << endl;
	cout << "***************************************************** " << endl;

	fTimeSec = 0;
	float fHashTime = 0;

	// Creating the offset-hash file
	CreateOffsetHash( m_wsRel1FileNameFlash, wsOffsetHashFile, 0.5f, nNumberOfWords, fHashTime);


	OffsetHashJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile, 0.5f,
					fTimeSec, nNumberOfWords, nReadSectorsNum);

	PrintResults(fResults, fTimeSec + fHashTime, nReadSectorsNum);


	cout << endl << "Performing sorted block join (No Rs  + out) " << endl;
	cout << "***************************************************** " << endl;
	
	fTimeSec = 0;
	fHashTime = 0;
#ifdef __FLASH__
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\E:\\Rel2Sorted", L"\\\\.\\E:\\rel2RS", false, 0.5f, fTimeSec, nNumberOfWords, nReadSectorsNum, eOuterSort);
#else
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\C:\\Rel2Sorted", L"\\\\.\\C:\\rel2RS", false, 0.1f, fTimeSec, nNumberOfWords, nReadSectorsNum, eOuterSort);
#endif // __FLASH__

	PrintResults(fResults, fTimeSec + fHashTime, nReadSectorsNum);



	cout << endl << "Performing sorted block join - (No Rs  + inner) " << endl;
	cout << "***************************************************** " << endl;
	
	fTimeSec = 0;
#ifdef __FLASH__ 
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\E:\\Rel2Sorted", L"\\\\.\\E:\\rel2RS", false, 0.5f, fTimeSec, nNumberOfWords, nReadSectorsNum, eInnerSort);
#else
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\C:\\Rel2Sorted", L"\\\\.\\C:\\rel2RS", false, 0.1f, fTimeSec, nNumberOfWords, nReadSectorsNum, eInnerSort);
#endif // __FLASH__
	PrintResults(fResults, fTimeSec + fHashTime, nReadSectorsNum);



	cout << endl << "Performing sorted block join (Rs  + No Sort) " << endl;
	cout << "***************************************************** " << endl;
	
	fTimeSec = 0;
#ifdef __FLASH__ 
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
		L"\\\\.\\E:\\Rel2Sorted", L"\\\\.\\E:\\rel2RS", true, 0.5f, fTimeSec, nNumberOfWords, nReadSectorsNum, eNoSort);
#else
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\C:\\Rel2Sorted", L"\\\\.\\C:\\rel2RS", true, 0.1f, fTimeSec, nNumberOfWords, nReadSectorsNum, eNoSort);
#endif // __FLASH__ 
	PrintResults(fResults, fTimeSec + fHashTime, nReadSectorsNum);

	cout << endl << "Performing sorted block join (Rs  + out) " << endl;
	cout << "***************************************************** " << endl;
	
	fTimeSec = 0;
#ifdef __FLASH__
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\E:\\Rel2Sorted", L"\\\\.\\E:\\rel2RS", true, 0.5f, fTimeSec, nNumberOfWords, nReadSectorsNum, eOuterSort);
#else
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\C:\\Rel2Sorted", L"\\\\.\\C:\\rel2RS", true, 0.1f, fTimeSec, nNumberOfWords, nReadSectorsNum, eOuterSort);
#endif // __FLASH__
	PrintResults(fResults, fTimeSec + fHashTime, nReadSectorsNum);

	cout << endl << "Performing sorted block join - (Rs  + inner) " << endl;
	cout << "***************************************************** " << endl;
	
	fTimeSec = 0;
#ifdef __FLASH__
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\E:\\Rel2Sorted", L"\\\\.\\E:\\rel2RS", true, 0.5f, fTimeSec, nNumberOfWords, nReadSectorsNum, eInnerSort);
#else
	SortedBlockJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, wsOffsetHashFile,
	L"\\\\.\\C:\\Rel2Sorted", L"\\\\.\\C:\\rel2RS", true, 0.1f, fTimeSec, nNumberOfWords, nReadSectorsNum, eInnerSort);
#endif // __FLASH__
	PrintResults(fResults, fTimeSec + fHashTime, nReadSectorsNum);



	cout << endl << "Performing local sorted block join " << endl;
	cout << "***************************************************** " << endl;
	fTimeSec = 0;
#ifdef __FLASH__
	SortedBlockLocalJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, L"\\\\.\\E:\\Hash1",
		L"\\\\.\\E:\\SortRel1",  L"\\\\.\\E:\\rel1RS", false, 0.5f, fTimeSec, nReadSectorsNum, eNoSort);
#else
	SortedBlockLocalJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, L"\\\\.\\C:\\Hash1",
							L"\\\\.\\C:\\SortRel1",  L"\\\\.\\C:\\rel1RS", false, 0.1f, fTimeSec, nReadSectorsNum, eNoSort);
#endif // __FLASH__
	PrintResults(fResults, fTimeSec, nReadSectorsNum);
	

	cout << endl << "Performing local sorted block join - Inner Sort" << endl;
	cout << "***************************************************** " << endl;
	fTimeSec = 0;
#ifdef __FLASH__
	SortedBlockLocalJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, L"\\\\.\\E:\\Hash1",
		L"\\\\.\\E:\\SortRel1",  L"\\\\.\\E:\\rel1RS", false, 0.5f, fTimeSec, nReadSectorsNum, eInnerSort);
#else
	SortedBlockLocalJoin(m_wsRel1FileNameFlash, m_wsRel2FileNameFlash, L"\\\\.\\C:\\Hash11",
							L"\\\\.\\C:\\SortRel11",  L"\\\\.\\C:\\rel1RS1", false, 0.1f, fTimeSec, nReadSectorsNum, eInnerSort);
#endif // __FLASH__
	PrintResults(fResults, fTimeSec, nReadSectorsNum);
	fResults << endl;


}
void JoinTests::PostTest()
{
}



float JoinTests::CalcTime(DWORD dwStart, DWORD dwStop)
{
	DWORD dwDiffMsec = (dwStop - dwStart);
	float fDiffSec = static_cast<float>(dwDiffMsec)  / (1000.0f);
#ifdef __FLASH_DEBUG__
	cout << "It took " << dwDiffMsec << " msec " << fDiffSec << " seconds - " << fDiffSec / 60.0f << " minutes" <<endl;
#endif __FLASH_DEBUG__
	//return fDiffSec;
	return static_cast<float>(dwDiffMsec);
}

bool JoinTests::CreateOffsetHash( wstring wsRel1FileName, wstring wsHashFileName, 
								 float fColumnCDensity, int &nHashWords, float& rfTimeSec)
{
	SRel1* psRel1Buffer			= NULL;
	int* pnColHash				= NULL;
	sRelResult *psResultBuffer	= NULL;

	if (AllocateVirtualBuffer((void**)(&psRel1Buffer), m_nRel1ByteCount ) == false)
	{
		return false;
	}

	Disk* psDiskRel1 = new Disk(wsRel1FileName, eLogical, eRead);
	Disk* psDiskHash = new Disk(wsHashFileName, eLogical, eWrite);

	SColHashElem* psColHash = (SColHashElem*)malloc(m_nRel1RowsNum * sizeof(SColHashElem));

	int nAllocNum = 500;

	if (psDiskRel1->OpenDisk() && psDiskHash->OpenDisk())
	{
		if (psDiskRel1->CreateOverlappedEvent() && 
			psDiskHash->CreateOverlappedEvent())
		{
			DWORD dwBytesRead = 0;
			int nResultCounter = 0;

			// Read the bigger relation and create the hash

			DWORDLONG dwlPrepareOffset = 0;
			int nMaxappearance = 0;

		
			// Initialize the buffer to 0
			memset ((void*)psColHash, 0, m_nRel1RowsNum * sizeof(SColHashElem));
			
			int nNumberOfWords = 0;
			
			DWORD dwStart = timeGetTime();
			// Read the bigger relation
			if (psDiskRel1->DoOverlappedIO(psRel1Buffer, m_nRel1ByteCount, dwlPrepareOffset))			
			{
				for (int i = 0; i < m_nRel1RowsNum; ++i)
				{
					int nIndex = psRel1Buffer[i].m_nc - 1; // shift to array start
					int nElemNum = psColHash[nIndex].nIndex;
					if(nElemNum == 0)
					{
						if ((psColHash[nIndex].pnOffset = (int*)malloc(nAllocNum * sizeof(int))) == NULL)
						{
							cout << "cant allocating" << endl;
						}
						nNumberOfWords += 2; // element + number of offsets
					}
					if (nElemNum < nAllocNum)
					{
						psColHash[nIndex].nElem = psRel1Buffer[i].m_nc;
						psColHash[nIndex].pnOffset[nElemNum] = i;// * sizeof(SRel2);
						psColHash[nIndex].nIndex += 1;
						if (nMaxappearance < psColHash[nIndex].nIndex)
						{
							nMaxappearance = psColHash[nIndex].nIndex;
						}
						nNumberOfWords++;
					}
					else
					{
						cout << "Problem!!!" << endl;
					}

				}

			}

			// transaction size must be aligned to 512 bytes
			if (nNumberOfWords % 512 != 0)
			{
				nNumberOfWords = ((nNumberOfWords / 512) + 1 ) * 512;
			}

			nHashWords = nNumberOfWords;

			int nColHashByteCount = nNumberOfWords *sizeof(int);
			if ( AllocateVirtualBuffer((void**)&(pnColHash), nColHashByteCount) == true)
			{
				int nHashElem = 0;

				// Create the hash
				for (int i = 0; i <  m_nRel1RowsNum * fColumnCDensity; ++i)
				{
					int nIndex= psColHash[i].nIndex;
					if(nIndex != 0)
					{
						pnColHash[nHashElem+0] = psColHash[i].nElem;
						pnColHash[nHashElem+1] = psColHash[i].nIndex;

						for (int j = 0; j < nIndex; ++ j)
						{
							pnColHash[nHashElem+2+j] = psColHash[i].pnOffset[j];
						}

						nHashElem += 2+nIndex;
						
						//#ifdef __FLASH_DEBUG__
						//if ( i >= 2 && i <= 10)
						//{
						//	cout << "C = " << psColHash[i].nElem << " : ";

						//	for (int j = 0; j < nIndex; ++j )
						//	{
						//		cout << psColHash[i].pnOffset[j] << ", ";
						//	}
						//	cout << endl;	
						//}
						//#endif // __FLASH_DEBUG__

					}
				}

				// Write the hash to disk

				if (psDiskHash->DoOverlappedIO(pnColHash,nColHashByteCount) == false)
				{
					cout << "Failed creating the hash-map file" << endl;
					return false;
				}

				DeAllocateVirtualBuffer((void**)(&pnColHash), nColHashByteCount);

			}

				DWORD dwStop = timeGetTime();
				rfTimeSec += CalcTime(dwStart, dwStop);
				cout << "Number Of Words in hash-map File " << nNumberOfWords << endl;
		}


		DeAllocateVirtualBuffer((void**)(&psRel1Buffer), m_nRel1ByteCount);


		psDiskHash->CloseDisk();
		psDiskRel1->CloseDisk();

		delete psDiskHash;
		delete psDiskRel1;
	}

	free(psColHash);
	return true;
}

bool JoinTests::OffsetHashJoin(	wstring wsRel1FileName, wstring wsRel2FileName,
								wstring wsHashFileName, float fColumnCDensity,
								float &rfTimeSec, int nNumberOfWords, int& rnReadSectorsNum)
{
	bool bRes					= false;
	SRel1* psRel1Buffer			= NULL;
	SRel2* psRel2Buffer			= NULL;
	sRelResult* psResultBuffer	= NULL;
	int* pnColHash				= NULL;
	int nResultCounter			= 0;
	DWORDLONG dwlPrepareOffset	= 0;
	rnReadSectorsNum			= 0;

	int nRel1IterationNumber = m_nRel1ByteCount / nBufferLimitedByteCount;
	int nRel2IterationNumber = m_nRel2ByteCount / nBufferLimitedByteCount;


	int nBufferRel1RowsNum = m_nRel1RowsNum / nRel1IterationNumber;
	int nBufferRel2RowsNum = m_nRel2RowsNum / nRel2IterationNumber;

	//int nBuffer1LimitedByteCount = 100*1024; // 100 KB
	//int nBuffer2LimitedByteCount = 100*1024; // 100 KB

	//int nBuffer1LimitedByteCount = (nBufferLimitedByteCount); // 100 KB
	//int nBuffer2LimitedByteCount = (nBufferLimitedByteCount); // 100 KB

	DWORDLONG dwlRel2Offset = 0;

	//int nNumberOfWords = 0;
	//#ifdef __CREATE_HASH__
	//	CreateOffsetHash( wsRel1FileName, wsHashFileName, fColumnCDensity, nNumberOfWords);
	//#else
	//	nNumberOfWords = 100864;

	//#endif // __CREATE_HASH__
	
	// Assumption: the hash size is smaller then the memory and whole of it can be read in one transaction
	int nColHashByteCount = nNumberOfWords *sizeof(int);

	if (AllocateVirtualBuffer((void**)(&pnColHash), nColHashByteCount) &&
		AllocateVirtualBuffer((void**)(&psRel1Buffer), nBufferLimitedByteCount) &&
		AllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount))
	{
		Disk* psDiskRel1 = new Disk(wsRel1FileName, eLogical, eRead);
		Disk* psDiskRel2 = new Disk(wsRel2FileName, eLogical, eRead);
		Disk* psDiskHash = new Disk(wsHashFileName, eLogical, eRead);
		
		int nResultsLimitWords = m_nRel1RowsNum * 10;
		// Allocate memory for the results
		psResultBuffer = (sRelResult*)malloc(nResultsLimitWords * sizeof(sRelResult));

		if (psDiskRel1->OpenDisk() && psDiskRel2->OpenDisk() && psDiskHash->OpenDisk())
		{
			if (psDiskRel1->CreateOverlappedEvent() && psDiskRel2->CreateOverlappedEvent() &&
				psDiskHash->CreateOverlappedEvent())
			{
				int nReadSectorNum = 0;

				DWORD dwStart = timeGetTime();

				// Reading the hash-table
				if (psDiskHash->DoOverlappedIO(pnColHash, nColHashByteCount))
				{
					// Succeeded in reading the hash-table
					nReadSectorNum += nColHashByteCount / 512;

					// Reading the smaller table

					for (int k = 0; k < nRel2IterationNumber; ++k)
					{
						if  (psDiskRel2->DoOverlappedIO(psRel2Buffer, nBufferLimitedByteCount, dwlRel2Offset))
						{
							// Succeeded in reading the smaller table
							nReadSectorNum += nBufferLimitedByteCount / 512;

							// Every word is 4 bytes
							int nWordsCount =  nColHashByteCount / 4;

							bool bRes = false;
							int nOffsetsNum = 0;
							int nHashElem = 0;
							int nElem = 0;
							int nOffsetIndex = 0;

							// For every word in the smaller table
							for (int i = 0 ; i < nBufferRel2RowsNum; ++ i)
							{
								
								bRes = FindElemInHash(psRel2Buffer, pnColHash, i, nWordsCount, 
									2, nOffsetsNum, nElem, nOffsetIndex);
								
								if (bRes)
								{
									//cout << "nHashElem = " << nHashElem << ", nOffsetsNum =  " << nOffsetsNum << endl;
									for (int j = 0; j < nOffsetsNum; ++j)
									{
										dwlPrepareOffset = pnColHash[nOffsetIndex + 2 + j] * 4 * 4; // 4 words in a row - 4 bytes per word

										// The offset address must be aligned to 512 - if not - take the 
										// start of the closest sector
										if ((dwlPrepareOffset % 512) != 0)
										{
											dwlPrepareOffset = (dwlPrepareOffset / 512) * 512;
										}

										int nLineNum = (pnColHash[nOffsetIndex + 2 + j] % 32); // 32 lines in a sector
										int nBuffer1LimitedByteCount = 512;
										nReadSectorNum++;

										
										// Read the fit element from the bigger table
										if  (psDiskRel1->DoOverlappedIO(psRel1Buffer, nBuffer1LimitedByteCount, dwlPrepareOffset))
										{
											if (psRel1Buffer[nLineNum].m_nc == psRel2Buffer[i].m_nc)
											{
												// There is a match - insert to the result buffer

												psResultBuffer[nResultCounter].CopyRel1(psRel1Buffer[nLineNum]);
												psResultBuffer[nResultCounter].CopyRel2(psRel2Buffer[i]);
												if (nResultsLimitWords == nResultCounter)
												{
													printf("%d, %d, %d\n", k,i,j);

												}
												nResultCounter++;

											}
											
										}
									}
								}
							}
							dwlRel2Offset += nBufferLimitedByteCount;
						}
					}
				}
				DWORD dwStop = timeGetTime();
				rfTimeSec += CalcTime(dwStart, dwStop);

				cout << "Number Of Rows in Result Join Relation " << nResultCounter << endl;
				cout << "Number Of Read Sectors are: " << nReadSectorNum << endl;

				
				rnReadSectorsNum = nReadSectorNum;
			}		
			else
			{
				bRes = false;
			}
		}
		else
		{
			bRes = false;
		}

		DeAllocateVirtualBuffer((void**)(&psRel1Buffer), nBufferLimitedByteCount);
		DeAllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount);
		free(psResultBuffer);

		psDiskHash->CloseDisk();
		psDiskRel1->CloseDisk();
		psDiskRel2->CloseDisk();

		delete psDiskHash;
		delete psDiskRel1;
		delete psDiskRel2;

	}
	else
	{
		bRes = false;
	}
	return bRes;

}

bool JoinTests::OffsetHashJoin1(wstring wsRel1FileName, wstring wsRel2FileName,
								wstring wsHashFileName, float fColumnCDensity,
								int nNumberOfWords, float& rfTimeSec, int& rnReadSectorsNum, ESortType bUseInnerSort)
{
	bool bRes					= false;
	SRel1* psRel1Buffer			= NULL;
	SRel2* psRel2Buffer			= NULL;
	sRelResult* psResultBuffer	= NULL;
	int* pnColHash				= NULL;
	int nResultCounter			= 0;
	DWORDLONG dwlPrepareOffset	= 0;
	DWORDLONG dwlRel2Offset		= 0;

	int nLastElem = 0;
	int nLastNumberOfMatches = 0;

	int nRel1IterationNumber = m_nRel1ByteCount / nBufferLimitedByteCount;
	int nRel2IterationNumber = m_nRel2ByteCount / nBufferLimitedByteCount;

	int nBufferRel1RowsNum = m_nRel1RowsNum / nRel1IterationNumber;
	int nBufferRel2RowsNum = m_nRel2RowsNum / nRel2IterationNumber;

	//int nBuffer1LimitedByteCount = 100*1024; // 100 KB
	//int nBuffer2LimitedByteCount = 100*1024; // 100 KB

	//int nBuffer1LimitedByteCount = MEMORY_SIZE_BYTES; // 100 KB
	//int nBuffer2LimitedByteCount = MEMORY_SIZE_BYTES; // 100 KB

	////int nNumberOfWords = 0;
	//#ifdef __CREATE_HASH__
	//	CreateOffsetHash( wsRel1FileName, wsHashFileName, fColumnCDensity, nNumberOfWords);
	//#else
	//	//nNumberOfWords = 100864;

	//#endif // __CREATE_HASH__
	
	int nColHashByteCount = nNumberOfWords *sizeof(int);

	if (AllocateVirtualBuffer((void**)(&pnColHash), nColHashByteCount) &&
		AllocateVirtualBuffer((void**)(&psRel1Buffer), nBufferLimitedByteCount) &&
		AllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount))
	{
		Disk* psDiskRel1 = new Disk(wsRel1FileName, eLogical, eRead);
		Disk* psDiskRel2 = new Disk(wsRel2FileName, eLogical, eRead);
		Disk* psDiskHash = new Disk(wsHashFileName, eLogical, eRead);
		
		VirtualLock((&pnColHash), nColHashByteCount);
		// Allocate memory for the results
		psResultBuffer = (sRelResult*)malloc(m_nRel1RowsNum * 10 * sizeof(sRelResult));

		if (psDiskRel1->OpenDisk() && psDiskRel2->OpenDisk() && psDiskHash->OpenDisk())
		{
			if (psDiskRel1->CreateOverlappedEvent() && psDiskRel2->CreateOverlappedEvent() &&
				psDiskHash->CreateOverlappedEvent())
			{
				int nReadSectorNum = 0;

				DWORD dwStart = timeGetTime();

				// Reading the hash-table
				if (psDiskHash->DoOverlappedIO(pnColHash, nColHashByteCount))
				{
					// Succeeded in reading the hash-table
					nReadSectorNum += nColHashByteCount / 512;

					// Reading the smaller table

					for (int k = 0; k < nRel2IterationNumber; ++k)
					{
						if  (psDiskRel2->DoOverlappedIO(psRel2Buffer, nBufferLimitedByteCount, dwlRel2Offset))
						{
							// Succeeded in reading the smaller table
							nReadSectorNum += nBufferLimitedByteCount / 512;

							// Every word is 4 bytes
							int nWordsCount =  nColHashByteCount / 4;

							bool bRes = false;
							int nOffsetsNum = 0;
							int nHashElem = 0;
							int nElem = 0;
							int nOffsetIndex = 0;


							if (eInnerSort == bUseInnerSort)
							{
								// Sort the smaller relation block
								m_pSortTests->QuickSort(psRel2Buffer, 0, nBufferRel2RowsNum);	
//								ListinsertSort(psRel2Buffer, nBufferRel2RowsNum);
							}

							// For every word in the smaller table
							for (int i = 0 ; i < nBufferRel2RowsNum; ++ i)
							{
								bRes = FindElemInHash(psRel2Buffer, pnColHash, i, nWordsCount, 
									2, nOffsetsNum, nElem, nOffsetIndex);

								if (bRes)
								{
									if (nLastElem != nElem)
									{
										// this is a new element which we didn't join it
										//cout << "nHashElem = " << nHashElem << ", nOffsetsNum =  " << nOffsetsNum << endl;
										for (int j = 0; j < nOffsetsNum; ++j)
										{
											dwlPrepareOffset = pnColHash[nOffsetIndex + 2 + j] * 4 * 4; // 4 words in a row - 4 bytes per word

											// The offset address must be aligned to 512 - if not - take the 
											// start of the closest sector
											if ((dwlPrepareOffset % 512) != 0)
											{
												dwlPrepareOffset = (dwlPrepareOffset / 512) * 512;
											}

											int nLineNum = (pnColHash[nOffsetIndex + 2 + j] % 32); // 32 lines in a sector
											int nBuffer1LimitedByteCount = 512;
											

											// Read the fit element from the bigger table
											if  (psDiskRel1->DoOverlappedIO(psRel1Buffer, nBuffer1LimitedByteCount, dwlPrepareOffset))
											{
												if (psRel1Buffer[nLineNum].m_nc == psRel2Buffer[i].m_nc)
												{
													// There is a match - insert to the result buffer
													psResultBuffer[nResultCounter].CopyRel1(psRel1Buffer[nLineNum]);
													psResultBuffer[nResultCounter].CopyRel2(psRel2Buffer[i]);
													nResultCounter++;
													nReadSectorNum ++;
												}
											}
										}
										nLastElem = nElem;
										nLastNumberOfMatches = nOffsetsNum;
									}
									else
									{
										// Due to the sorted file - we don't need to read the buffer 1 lines
										// but to copy them
										
										int nIndex = nResultCounter - nLastNumberOfMatches;
										for (int j = 0; j < nLastNumberOfMatches; ++j)
										{

											psResultBuffer[nResultCounter].CopyRel1(psResultBuffer[nIndex].m_sRel1);
											psResultBuffer[nResultCounter].CopyRel2(psRel2Buffer[i]);
											nResultCounter++;
											nIndex++;
										}
									}

								}
							}
							dwlRel2Offset += nBufferLimitedByteCount;
						}
					}
				}
				DWORD dwStop = timeGetTime();
				rfTimeSec += CalcTime(dwStart, dwStop);

				rnReadSectorsNum += nReadSectorNum;

				cout << "Number Of Rows in Result Join Relation " << nResultCounter << endl;
				cout << "Number Of Read Sectors are: " << rnReadSectorsNum << endl;

				
			}		
			else
			{
				bRes = false;
			}
		}
		else
		{
			bRes = false;
		}

		VirtualUnlock((&pnColHash), nColHashByteCount);

		DeAllocateVirtualBuffer((void**)(&psRel1Buffer), nBufferLimitedByteCount);
		DeAllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount);
		DeAllocateVirtualBuffer((void**)(&pnColHash), nColHashByteCount);
		free(psResultBuffer);

		psDiskHash->CloseDisk();
		psDiskRel1->CloseDisk();
		psDiskRel2->CloseDisk();

		delete psDiskHash;
		delete psDiskRel1;
		delete psDiskRel2;
	}
	else
	{
		bRes = false;
	}
	return bRes;

}

bool JoinTests::OffsetHashJoin2(wstring wsRel1FileName, wstring wsRel2FileName,
								wstring wsHashFileName, float fColumnCDensity,
								int nNumberOfWords, float& rfTimeSec,  int& rnReadSectorsNum, ESortType eSortType)
{
	bool bRes					= false;
	SRel1* psRel1Buffer			= NULL;
	SRel2* psRel2Buffer			= NULL;
	sRelResult* psResultBuffer	= NULL;
	int* pnColHash				= NULL;
	int nResultCounter			= 0;
	DWORDLONG dwlPrepareOffset	= 0;
	DWORDLONG dwlLastOffset		= 0;
	DWORDLONG dwlRel2Offset		= 0;
	rnReadSectorsNum			= 0;

	int nRel1IterationNumber = m_nRel1ByteCount / nBufferLimitedByteCount;
	int nRel2IterationNumber = m_nRel2ByteCount / nBufferLimitedByteCount;

	int nBufferRel1RowsNum = m_nRel1RowsNum / nRel1IterationNumber;
	int nBufferRel2RowsNum = m_nRel2RowsNum / nRel2IterationNumber;

	//int nBuffer1LimitedByteCount = 100*1024; // 100 KB
	//int nBuffer2LimitedByteCount = 100*1024; // 100 KB

	int nColHashByteCount = nNumberOfWords *sizeof(int);

	if (AllocateVirtualBuffer((void**)(&pnColHash), nColHashByteCount) &&
		AllocateVirtualBuffer((void**)(&psRel1Buffer), nBufferLimitedByteCount) &&
		AllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount))
	{
		Disk* psDiskRel1 = new Disk(wsRel1FileName, eLogical, eRead);
		Disk* psDiskRel2 = new Disk(wsRel2FileName, eLogical, eRead);
		Disk* psDiskHash = new Disk(wsHashFileName, eLogical, eRead);
		
		// Allocate memory for the results
		psResultBuffer = (sRelResult*)malloc(m_nRel1RowsNum * 10 * sizeof(sRelResult));

		if (psDiskRel1->OpenDisk() && psDiskRel2->OpenDisk() && psDiskHash->OpenDisk())
		{
			if (psDiskRel1->CreateOverlappedEvent() && psDiskRel2->CreateOverlappedEvent() &&
				psDiskHash->CreateOverlappedEvent())
			{
				int nReadSectorNum = 0;

				DWORD dwStart = timeGetTime();

				// Reading the hash-table
				if (psDiskHash->DoOverlappedIO(pnColHash, nColHashByteCount))
				{
					// Succeeded in reading the hash-table
					nReadSectorNum += nColHashByteCount / 512;

					// Reading the smaller table
					for (int k = 0; k < nRel2IterationNumber; ++k)
					{

						if  (psDiskRel2->DoOverlappedIO(psRel2Buffer, nBufferLimitedByteCount, dwlRel2Offset))
						{
							// Succeeded in reading the smaller table
							nReadSectorNum += nBufferLimitedByteCount / 512;

							// Every word is 4 bytes
							int nWordsCount =  nColHashByteCount / 4;
							
							int nNumOfLinesInBlock	= (nBufferLimitedByteCount / 16);

							bool bRes = false;
							int nOffsetsNum = 0;
							int nHashElem = 0;
							int nElem = 0;
							int nOffsetIndex = 0;

							SRel2* psRel2Copy = (SRel2*)malloc(nBufferLimitedByteCount);
							
							if (eInnerSort == eSortType)
							{
								// Sort the block
								m_pSortTests->QuickSort(psRel2Buffer, 0, nBufferRel2RowsNum);
							}

							// For every word in the smaller table
							for (int i = 0 ; i < nBufferRel2RowsNum; ++ i)
							{
								bRes = FindElemInHash(psRel2Buffer, pnColHash, i, nWordsCount, 
									2, nOffsetsNum, nElem, nOffsetIndex);

								cout << psRel2Buffer[i].GetKey() << "found " << nOffsetsNum << " matches" << endl;

								if (bRes)
								{
									// this is a new element which we didn't join it
									//cout << "nHashElem = " << nHashElem << ", nOffsetsNum =  " << nOffsetsNum << endl;
									for (int j = 0; j < nOffsetsNum; ++j)
									{
										dwlPrepareOffset = pnColHash[nOffsetIndex + 2 + j] * 4 * 4; // 4 words in a row - 4 bytes per word

										// The offset address must be aligned to 512 - if not - take the 
										// start of the closest sector
										if ((dwlPrepareOffset % 512) != 0)
										{
											dwlPrepareOffset = (dwlPrepareOffset / 512) * 512;
										}

										int nLineNum = (pnColHash[nOffsetIndex + 2 + j] % 32); // 32 lines in a sector
										if (dwlPrepareOffset == dwlLastOffset && (0!=i && 0!=j))
										{
											// Due to the sorted file there is a possibility
											// the last sector was as the current - so we don't
											// have to read it again

											if (psRel1Buffer[nLineNum].m_nc == psRel2Buffer[i].m_nc)
											{
												// There is a match - insert to the result buffer
												psResultBuffer[nResultCounter].CopyRel1(psRel1Buffer[nLineNum]);
												psResultBuffer[nResultCounter].CopyRel2(psRel2Buffer[i]);
												nResultCounter++;
											}
											else
											{
												printf("problem\n");
											}
										}
										else
										{							
											int nBuffer1LimitedByteCount = 512;
											

											// Read the fit element from the bigger table
											if  (psDiskRel1->DoOverlappedIO(psRel1Buffer, nBuffer1LimitedByteCount, dwlPrepareOffset))
											{
												if (psRel1Buffer[nLineNum].m_nc == psRel2Buffer[i].m_nc)
												{
													// There is a match - insert to the result buffer
													psResultBuffer[nResultCounter].CopyRel1(psRel1Buffer[nLineNum]);
													psResultBuffer[nResultCounter].CopyRel2(psRel2Buffer[i]);
													nResultCounter++;
														
													dwlLastOffset = dwlPrepareOffset;
													nReadSectorNum++;

												}
											}
										}
									}
								}

							}
						}
						dwlRel2Offset += nBufferLimitedByteCount;
					}
				}
				DWORD dwStop = timeGetTime();
				rfTimeSec += CalcTime(dwStart, dwStop);

				rnReadSectorsNum += nReadSectorNum;

				cout << "Number Of Rows in Result Join Relation " << nResultCounter << endl;
				cout << "Number Of Read Sectors are: " << rnReadSectorsNum << endl;
			}		
			else
			{
				bRes = false;
			}
		}
		else
		{
			bRes = false;
		}

		DeAllocateVirtualBuffer((void**)(&psRel1Buffer), nBufferLimitedByteCount);
		DeAllocateVirtualBuffer((void**)(&psRel2Buffer), nBufferLimitedByteCount);
		free(psResultBuffer);

		psDiskHash->CloseDisk();
		psDiskRel1->CloseDisk();
		psDiskRel2->CloseDisk();

		delete psDiskHash;
		delete psDiskRel1;
		delete psDiskRel2;
	}
	else
	{
		bRes = false;
	}
	return bRes;

}


bool JoinTests::SortedBlockJoin(wstring wsRel1FileName, wstring wsRel2FileName,
								wstring wsHashFileName, wstring wsSortedFileName, wstring wsRsFileName, bool bUseRs,
								float fColumnCDensity, float& rfTimeSec, int nNumOfWords,
								int& rnReadSectorsNum, ESortType eSortType)
{
	// TODO: remove wsSortedFileName from arguments list
	rnReadSectorsNum = 0;

	// TOOD: change to argument
	int nWrittemSectorsNum = 0;

	wstring wsRelFileName = wsRel2FileName;
	rfTimeSec = 0;

	if (true == bUseRs)
	{
		// Perform replacement selection to the big relation before performing the hash-join
		PerformRS<SRel2>( wsRel2FileName, wsRsFileName, m_nRel2RowsNum, m_nRel2ByteCount, rfTimeSec, rnReadSectorsNum);
		wsRelFileName = wsRsFileName;
		printf("the RS took : %f msec\n", rfTimeSec);
		printf("after RS num of read sectors is: %d \n", rnReadSectorsNum);
	}
	

	if (eOuterSort == eSortType)
	{
#ifndef __ITERATIONS_SORT__
		// Read the small relation - block by block
		// Sort every block and write it to another place in the flash
		m_pSortTests->CreateBlockSortedRelation<SRel2>(wsRelFileName, wsSortedFileName, wsRsFileName, m_nRel2RowsNum, m_nRel2ByteCount, 
			rfTimeSec, NULL, rnReadSectorsNum, nWrittemSectorsNum, bUseRs);
#else
		DWORD dwStart = timeGetTime();
		m_pSortTests->SortInIterations<SRel2>(0, 2, m_wsRel2FileNameFlash, L"\\\\.\\C:\\rel2Sort", m_nRel2RowsNum,
			m_nRel2ByteCount,rfTimeSec,rnReadSectorsNum, nWrittemSectorsNum, wsRelFileName);
		DWORD dwStop = timeGetTime();
		rfTimeSec += CalcTime(dwStart, dwStop);
#endif // __ITERATIONS_SORT__

//		wsRelFileName = wsSortedFileName;
		printf("after eOuterSort num of read sectors is: %d \n", rnReadSectorsNum);
	}
	
	// Read every block from the sorted file in flash
	// Perform the join - for every number which is the same there is no need to read the whole sector again
	OffsetHashJoin1(wsRel1FileName, wsRelFileName,
					wsHashFileName, fColumnCDensity,
					nNumOfWords, rfTimeSec, rnReadSectorsNum, eSortType);
	printf("after join num of read sectors is: %d \n", rnReadSectorsNum);

	return true;

}

bool JoinTests::SortedBlockLocalJoin(wstring wsRel1FileName, wstring wsRel2FileName,
								wstring wsHashFileName, wstring wsSortedFileName, wstring wsRsFileName, bool bUseRs,
								float fColumnCDensity, float& rfTimeSec, int& rnReadSectorsNum, ESortType eSortType)
{
	rfTimeSec = 0;
	rnReadSectorsNum = 0;

	// TODO: change to argument
	int nWrittemSectorsNum = 0;
	
	wstring wsRelFileName = wsRel1FileName;

	if (true == bUseRs)
	{
		// Perform replacement selection to the big relation before performing the hash-join
		PerformRS<SRel1>( wsRel1FileName, wsRsFileName, m_nRel1RowsNum, m_nRel1ByteCount, rfTimeSec, rnReadSectorsNum);
		wsRelFileName = wsRsFileName;
		printf("number of read sectors = %d\n", rnReadSectorsNum);
	}
	// Read the big relation block by block and sort them. Write the sorted
	// blocks to another file.
	wstring wsFinalSortFile;
#ifndef __ITERATIONS_SORT__
	m_pSortTests->CreateBlockSortedRelation<SRel1>(wsRelFileName, wsSortedFileName, wsRsFileName, m_nRel1RowsNum, m_nRel1ByteCount, rfTimeSec, NULL, rnReadSectorsNum, nWrittemSectorsNum, bUseRs);
	wsFinalSortFile = wsSortedFileName;
#else

	DWORD dwStart = timeGetTime();
	m_pSortTests->SortInIterations<SRel1>(0, 2, wsRelFileName, wsSortedFileName, m_nRel1RowsNum,
		m_nRel1ByteCount, rfTimeSec,rnReadSectorsNum, nWrittemSectorsNum, wsFinalSortFile);

	DWORD dwStop = timeGetTime();
	rfTimeSec += CalcTime(dwStart, dwStop);
#endif //  __ITERATIONS_SORT__
	
	printf("number of read sectors = %d\n", rnReadSectorsNum);
	// Prepare the hash-file to the sorted relation.
	int nHashWordsNumber = 0;
	CreateOffsetHash(wsFinalSortFile/*wsSortedFileName*/, wsHashFileName, fColumnCDensity, nHashWordsNumber, rfTimeSec);

	printf("number of read sectors = %d\n", rnReadSectorsNum);
	// Perform the join and for every element read the needed sectors - hope for
	// locality and to read less blocks.

	OffsetHashJoin2(wsFinalSortFile/*wsSortedFileName*/, wsRel2FileName, wsHashFileName, 
					fColumnCDensity, nHashWordsNumber, rfTimeSec,  rnReadSectorsNum, eSortType);

	printf("number of read sectors = %d\n", rnReadSectorsNum);
	return true;
}

//template <class T>
//bool JoinTests::CreateBlockSortedRelation( wstring wsRelFileName, wstring wsSortedFileName, wstring wsRsFileName, int nRelRowsNum,
//											int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum, 
//											bool bUseRs)
//{
//	// TODO: don't need so many buffers - only one
//	bool bRes					= false;
//	T* psRelBuffer				= NULL;
//	T* psRelSoretdBuffer		= NULL;
//
//	if (AllocateVirtualBuffer((void**)(&psRelBuffer), nBufferLimitedByteCount ) == true)
//	{
//		if (AllocateVirtualBuffer((void**)(&psRelSoretdBuffer), nBufferLimitedByteCount ) == false)
//		{
//			DeAllocateVirtualBuffer((void**)(&psRelBuffer));
//		}
//	}
//	else
//	{
//		return false;
//	}
//
//	VirtualLock((void**)(&psRelBuffer), nBufferLimitedByteCount);
//	VirtualLock((void**)(&psRelSoretdBuffer), nBufferLimitedByteCount);
//
//	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
//	Disk* psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);
//
//	int nNumOfLinesInBlock	= (nBufferLimitedByteCount / sizeof(T));
//	int nNumOfIterations	= static_cast<int>(ceil(static_cast<float>(nRelByteCount) / nBufferLimitedByteCount));
//
//	if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
//	{
//		if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
//		{
//			DWORDLONG dwlOffset = 0;
//
//			DWORD dwStart = timeGetTime();
//
//			for (int j = 0; j < nNumOfIterations; ++j)
//			{
//				dwlOffset = (j * nBufferLimitedByteCount);
//				// TODO: add a case when the offset is not sector aligned
//				// Read a block of the relation
//				if (psDiskRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
//				{
//					////rnReadSectorsNum += nBufferLimitedByteCount / 512;
//					// Perform merge sort
//					//MergeSort(0, nNumOfLinesInBlock - 1, psRelBuffer, psRelSoretdBuffer);
//					m_pSortTests->QuickSort(psRelBuffer, 0, nNumOfLinesInBlock);
//	//				ListinsertSort(psRelBuffer, nNumOfLinesInBlock);
//
//					//for (int i = 0; i < nNumOfLinesInBlock; ++ i)
//					//{
//					//	psRelSoretdBuffer[i] = psRelBuffer[i];
//					//}
//
//					// Write the sorted buffer into the disk
//					if (psDiskSortedRel->DoOverlappedIO(psRelBuffer/*psRelSoretdBuffer*/, nBufferLimitedByteCount, dwlOffset, &rnWrittenSectorsNum))
//					{
//						bRes = true;
//						
//						//rnWrittenSectorsNum += nBufferLimitedByteCount / 512;
//					}
//					else
//					{
//						bRes = false;
//					}
//
//					//for (int i = 0; i < nNumOfLinesInBlock; ++i)
//					//{
//					//	cout << psRel1SoretdBuffer[i].m_nc << endl;
//					//}
//				}
//			}
//
//			DWORD dwStop = timeGetTime();
//			rfTimeSec += CalcTime(dwStart, dwStop);
//
//			psDiskRel->CloseDisk();
//			psDiskSortedRel->CloseDisk();
//		}
//	}
//
//	VirtualUnlock((void**)(&psRelBuffer), nBufferLimitedByteCount);
//	VirtualUnlock((void**)(&psRelSoretdBuffer), nBufferLimitedByteCount);
//
//	DeAllocateVirtualBuffer((void**)(&psRelBuffer));
//	DeAllocateVirtualBuffer((void**)(&psRelSoretdBuffer));
//
//	delete psDiskRel;
//	delete psDiskSortedRel;
//
//	return bRes;
//}



template <class T>
void JoinTests::Merge(int nLow,int nMid,int nHigh, T* psRel, T* psRelSorted)
{
	int h,i,j, k;
	h = nLow;
	i = nLow;
	j = nMid+1;

	 while((h <= nMid)&&(j <= nHigh))
	 {
		 if(psRel[h] <= psRel[j])
		 {
			psRelSorted[i] = psRel[h];
			h++;
		 }
		else
		{
			psRelSorted[i] = psRel[j];
			j++;
		}
		i++;
	}
	 if(h > nMid)
	 {
	  for(k = j; k <= nHigh; k++)
	  {
		  	psRelSorted[i] = psRel[k];
			i++;
	  }
	 }
	 else
	 {
	  for(k = h;k <= nMid; k++)
	  {
		  	psRelSorted[i] = psRel[k];
			i++;
	  }
	 }

	  for (k = nLow; k <= nHigh; k++)
	  {
			psRel[k] = psRelSorted[k];
	  }
}

template <class T>
void JoinTests::MergeSort(int nLow,int nHigh, T* psRel, T* psRelSorted)
{
	int nMid;
	if(nLow < nHigh)
	{
		nMid=(nLow + nHigh)/2;
		MergeSort(nLow, nMid, psRel, psRelSorted);
		MergeSort(nMid + 1, nHigh, psRel, psRelSorted);
		Merge(nLow, nMid, nHigh, psRel, psRelSorted);
	}
}

template <class T>
bool JoinTests::FindElemInHash(T* psRelBuffer, int* pnColHash, int nIndex, int nWordsCount, int nHeaderSize,
							   int& nOffsetsNum, int& nElem, int& nOffsetIndex)
{
	bool bRes			= false;
	int nHashElem		= 0;
	nElem				= psRelBuffer[nIndex].GetKey();

	// find the element in the hash table
	for (int j = 0; j < nWordsCount-1;)
	{
		nHashElem			= pnColHash[j];
		nOffsetsNum			= pnColHash[j+1];
		nOffsetIndex		= j;

		if (nHashElem == nElem)
		{
			// we found the element we wanted
			bRes = true;
			break;
		}
		else if (nHashElem < nElem)
		{
			// The elements are sorted to proceed to the next element
			j += nHeaderSize + nOffsetsNum;
		}
		else if (nHashElem > nElem)
		{
			// Due to the order - the element was not found and we
			// can stop searching
			break;
		}
	}

	return bRes;
}

// Find the smaller element in a cyclic buffer (in range of nWordsCount words)
template <class T>
void JoinTests::FindSmallerElement(T* psRelBuffer, int nIndex, int nWordsCount, int nTotalWordsCount, T lastElement)
{
	// In the end - the element in place nIndex will be the smaller that is bigger than the last smaller one

	bool bRes			= false;
	int nSmallestIndex	= nIndex;

	for (int i = 0; i < nWordsCount; ++i)
	{
		int nOffset = i + nIndex;
		if (nOffset >= nTotalWordsCount)
		{
			nOffset = nOffset % nWordsCount;
		}

		if (psRelBuffer[nOffset] <= psRelBuffer[nIndex] && lastElement <= psRelBuffer[nOffset])
		{
			// Switch between the two elements
			T temp = psRelBuffer[nIndex];
			psRelBuffer[nIndex] = psRelBuffer[nOffset] ;
			psRelBuffer[nOffset] = temp;	
			bRes = true;
		}else if (psRelBuffer[nOffset] < psRelBuffer[nIndex])
		{
			nSmallestIndex = nOffset;
		}

	}

	if (false == bRes)
	{
		// we didn't find the next element in the same run - 
		// So we need to start a new run

		// We already find the smallest elenent - we will switch between
		// the first one and the smallest

		T temp = psRelBuffer[nIndex];
		psRelBuffer[nIndex] = psRelBuffer[nSmallestIndex] ;
		psRelBuffer[nSmallestIndex] = temp;	

		//printf("\n");
	}

}

template <class T>
void JoinTests::PerformRS( wstring wsRelFileName, wstring wsRSFileName,
							int nRelRowsNum, int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum)
{
	EChosenRsGroup eChosenGroup		= eNone;
	DWORDLONG dwlOffset				= 0;
	DWORDLONG dwlRSOffset			= 0;
	int nHalfBufferSizeByteCount	= nBufferLimitedByteCount / 2;
	int nBufferWordsCount			= nBufferLimitedByteCount / sizeof(T);
	int nHalfBufferWordsCount		= nHalfBufferSizeByteCount / sizeof(T);

	// TODO: handle with not whole number

	T* psRelBuffer				= NULL;
	T* psTempRelBuffer			= NULL;
	T* psRelRSdBuffer			= NULL;

	if (AllocateVirtualBuffer((void**)(&psRelBuffer), nBufferLimitedByteCount ) == true)
	{
		if (AllocateVirtualBuffer((void**)(&psRelRSdBuffer), nBufferLimitedByteCount ) == false)
		{
			DeAllocateVirtualBuffer((void**)(&psRelBuffer), nBufferLimitedByteCount);
		}
	}
	else
	{
		// TODO: handle with this
		return;
	}

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	Disk* psDiskRSdRel		= new Disk(wsRSFileName, eLogical, eWrite);

	int nNumOfLinesInBlock	= (nBufferLimitedByteCount / sizeof(T));

	if (psDiskRel->OpenDisk() && psDiskRSdRel->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskRSdRel->CreateOverlappedEvent())
		{
			DWORD dwStart = timeGetTime();

			int nRSIndex	= 0;
			int nIndex		= 0;
			T lastSmallerElement;

			// First,  read one buffer
			psDiskRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, dwlOffset);
			dwlOffset += nBufferLimitedByteCount;
			rnReadSectorsNum += nBufferLimitedByteCount / 512;
			int nSortedElementsNum = nHalfBufferWordsCount;

			while (nRelByteCount >= dwlOffset && nRelByteCount > dwlRSOffset)
			{
				// Find the next smaller element which is bigger than the last one
				// If there isn't one like this - find the smallest one
				FindSmallerElement(psRelBuffer, nIndex, nSortedElementsNum, nBufferWordsCount, lastSmallerElement);

				lastSmallerElement = psRelBuffer[nIndex];
				nIndex++;

				// Copy the element to the results buffer

				psRelRSdBuffer[nRSIndex++] = lastSmallerElement;
//				printf("%d, ", lastSmallerElement.GetKey());

				// Check if the results buffer is full, if so write it into the disk

				if (nRSIndex == nBufferWordsCount)
				{
					nRSIndex = 0;
					psDiskRSdRel->DoOverlappedIO(psRelRSdBuffer, nBufferLimitedByteCount, dwlRSOffset);
					dwlRSOffset += nBufferLimitedByteCount;
					// Count also the written sectors
					rnReadSectorsNum += nBufferLimitedByteCount / 512;

				}	

				// Check if the one half of the buffer was read completely
				if (nIndex == nHalfBufferWordsCount && nRelByteCount > dwlOffset)
				{
					// Read a new half buffer (into the first half of the buffer)
					psDiskRel->DoOverlappedIO(psRelBuffer, nHalfBufferSizeByteCount, dwlOffset);
					dwlOffset += nHalfBufferSizeByteCount;
					rnReadSectorsNum += nHalfBufferSizeByteCount / 512;
				}
				else if (nIndex == nBufferWordsCount && nRelByteCount > dwlOffset)
				{
					nIndex = 0;
					// Read a new half buffer (into the second half of the buffer)
					psDiskRel->DoOverlappedIO(psRelBuffer + nHalfBufferWordsCount, nHalfBufferSizeByteCount, dwlOffset);
					dwlOffset += nHalfBufferSizeByteCount;
					rnReadSectorsNum += nHalfBufferSizeByteCount / 512;
				}

				if (nRelByteCount == dwlOffset)
				{
					// On the last half-buffer we will get smaller elements
					// only from the rest of the half-buffer and not from all
					// of it
					nSortedElementsNum--;
				}

			}
			
			DWORD dwStop = timeGetTime();
			rfTimeSec += CalcTime(dwStart, dwStop);
		}

	}

	psDiskRel->CloseDisk();
	psDiskRSdRel->CloseDisk();

	DeAllocateVirtualBuffer((void**)(&psRelBuffer), nBufferLimitedByteCount);
	DeAllocateVirtualBuffer((void**)(&psRelRSdBuffer), nBufferLimitedByteCount);

	delete psDiskRel;
	delete psDiskRSdRel;

}


//void JoinTests::CheckForRunEnd(int& rnRSIndex, int nNumOfLinesInBlock, DWORDLONG& rdwlRSOffset, void* psRelRSdBuffer, Disk* psDiskRSdRel, bool bForceWrite)
//{
//	if (rnRSIndex == nNumOfLinesInBlock || true == bForceWrite)
//	{
//		// write the runs into the disk
//		if (psDiskRSdRel->DoOverlappedIO(psRelRSdBuffer, nBufferLimitedByteCount, rdwlRSOffset))
//		{
//			rnRSIndex = 0;
//			rdwlRSOffset += nBufferLimitedByteCount;
////			printf("\n **************** Wrote to disk - the buffer is full **************** \n");
//
//		}
//		else
//		{
//			printf("ERROR: cannot write the runs into the disk \n");
//		}
//
//	}
//}

void JoinTests::CheckForBufferRead(int& rnIndex, int nNumOfLinesInBlock, DWORDLONG& rdwlOffset, Disk* psDiskRel, void* psRelBuffer, DWORDLONG dwlBarrierOffset)
{
	if (rnIndex == nNumOfLinesInBlock)
	{
		if (rdwlOffset < dwlBarrierOffset)
		{
			// We need to read another part of the buffer
			if (psDiskRel->DoOverlappedIO(psRelBuffer, nBufferLimitedByteCount, rdwlOffset))
			{
				rnIndex = 0;
	//			printf("\n *************************** Read Another Buffer **********************\n");

			}
			else
			{
				// TODO: handle the last read - there is no need for it
				printf("ERROR - cannot read from buffer \n");
			}
		}
		rdwlOffset += nBufferLimitedByteCount;
	}
}


//template <class T>
//void JoinTests::QuickSort(T arr[], int nBegInd, int nEndInd)
//{
//  if (nEndInd > nBegInd + 1)
//  {
//    T piv			= arr[nBegInd];
//	int nleftInd	= nBegInd + 1;
//	int nRightInd	= nEndInd;
//
//    while (nleftInd < nRightInd)
//    {
//      if (arr[nleftInd] <= piv)
//        nleftInd++;
//      else
//	  {
//		// Swap
//		T temp			= arr[nleftInd];
//		arr[nleftInd]	= arr[--nRightInd];
//		arr[nRightInd]	= temp;
//	  }
//    }
//
//	// Swap
//	T temp			= arr[--nleftInd];
//	arr[nleftInd]	= arr[nBegInd];
//	arr[nBegInd]	= temp;
//
//    QuickSort(arr, nBegInd, nleftInd);
//    QuickSort(arr, nRightInd, nEndInd);
//  }
//}


template<class T> 
void JoinTests::ListinsertSort(T * v, int n)
 {
        int * link = new int[n];
        int head =0;
        int  next,cur,i;
        link[0] = -1;
        for (i = 1 ; i < n; i++){
               if (v[head] > v[i]){
                   link[i] = head;
                   head = i;
                } else {
                       for (cur = head ; cur != -1 && v[cur] <= v[i]; cur = link[cur])
                              next =cur;
                       link[next] = i;
                       link[i] = cur;
                }
       }
        cur = head;
        for ( i = 0 ;i < n;i++){
                while(cur < i)
                     cur = link [cur];
                next = link[cur];
                if (cur != i) {
                    T swap = v[i];
                    v[i] = v[cur];
                    v[cur] = swap;
                    link[cur] = link[i];
                    link[i] = cur;
                }
                cur = next;
        }
        delete[] link;
 }

void JoinTests::Test3(ostream& fResults, int nIteration)
{
	if (0 == nIteration)
	{
		fResults << "Memory Size[B] \t | Rel2   \t | Rel1 \t | Time [sec]   \t | IO Time [sec]   \t |NumOfReadSectors   \t | NumOfWrittenSectors   \t |  "<< endl;
		fResults << "--------------------------------------------------------------------------------------------------------------------------" << endl;
	}

	wstring wappends[] = {L"1", L"2", L"3", L"4", L"5",  L"6",L"1", L"2", L"3", L"4", L"5",  L"6"};

	wstring wsFinalFileName;

	float fTimeSec			= 0;
	float fTime				= 0;
	int nReadSectorsNum		= 0;
	int nWrittenSectorsNum	= 0;
	DWORD dwStart			= 0;
	DWORD dwStop			= 0;

	m_pSortTests->SetBufferLimitedByteCount(nBufferLimitedByteCount);

#ifdef __SORTED_PIECES__

		CreateRandomRelations2(0.1f, 1.0f, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
			m_wsRel1FileNameFlash, m_wsRel2FileNameFlash,5);

#else
	// Create the relations
	CreateRandomRelations(0.1f, 1.0f, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
		m_wsRel1FileNameFlash, m_wsRel2FileNameFlash);

#endif // __SORTED_PIECES__

	m_pSortTests->Test(fResults, nIteration, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
							nSmallRelByteCount, nBigRelByteCount, m_wsRel2FileNameFlash, wappends[nIteration]);

	//m_pSortTests->Test1(fResults, nIteration, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
	//						nSmallRelByteCount, nBigRelByteCount, m_wsRel2FileNameFlash);


	fResults << endl;


}

void JoinTests::SortingUnitTest()
{
	float fTime			= 0;
	int nReadSectorsNum = 0;

	m_pSortTests->SetBufferLimitedByteCount(nBufferLimitedByteCount);


#ifdef __SORTED_PIECES__
	m_pSortTests->PiecesUnitTests(m_nRel2ByteCount);
	return;
#else
	// Create the relations
	CreateRandomRelations(0.1f, 1.0f, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
							m_wsRel1FileNameFlash, m_wsRel2FileNameFlash);
	
	//CreateRandomRelations2(0.1f, 1.0f, m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount,
	//	m_wsRel1FileNameFlash, m_wsRel2FileNameFlash);



	m_pSortTests->SortingUnitTest1(m_nRel1RowsNum, m_nRel2RowsNum, m_nRel1ByteCount, m_nRel2ByteCount, m_wsRel1FileNameFlash, m_wsRel2FileNameFlash);

#endif // __SORTED_PIECES__
}
